Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c12c155dd8 | ||
|
|
c01d33f6e8 | ||
|
|
edc20b9e42 | ||
|
|
a3ae72d353 | ||
|
|
dc99f1fac3 |
No files matched your search
@@ -89,6 +89,7 @@ jobs:
|
||||
src/rune/canary_p0_04.py \
|
||||
src/rune/canary_p0_05.py \
|
||||
src/rune/canary_p0_07.py \
|
||||
src/rune/canary_p1_07.py \
|
||||
src/rune/canary_query_memory.py \
|
||||
src/rune/canary_unbounded.py \
|
||||
src/rune/canary_full_system.py \
|
||||
|
||||
@@ -401,6 +401,24 @@
|
||||
"role": "Literal chat presentation primitive",
|
||||
"disposition": "retain-authority",
|
||||
"evidence": "Renders role-specific text without evaluating message markup."
|
||||
},
|
||||
{
|
||||
"path": "src/rune/tool_protocol.py",
|
||||
"role": "Typed runtime-object set (MemoryQuery, Observation, VerificationResult, Outcome) with to_dict/from_dict round-trips",
|
||||
"disposition": "retain-authority",
|
||||
"evidence": "P1-07 (#30): frozen dataclasses carrying observable plan/action/result state on the response envelope; OperatorResponse.runtime_objects is the additive envelope key."
|
||||
},
|
||||
{
|
||||
"path": "src/rune/rune_tui.py",
|
||||
"role": "WorkPane operational index (third collapsible column, Ctrl+3) fed by _consume_runtime_objects",
|
||||
"disposition": "migrate-adapter",
|
||||
"evidence": "P1-07 (#30): projects runtime objects by stable id (update-in-place, newest-first sections); the chat log keeps its inline ToolCallWidget narrative (two-view principle)."
|
||||
},
|
||||
{
|
||||
"path": "src/rune/rune_operator.py",
|
||||
"role": "Runtime-object emission points (MemoryQuery from the broker trace, Observations, VerificationResult + Outcome) on the final response path",
|
||||
"disposition": "migrate-adapter",
|
||||
"evidence": "P1-07 (#30): built from _last_trace + the response; Outcome.status is back-filled from the final task transition; #33 (OutcomeLedger) adopts the Outcome shape."
|
||||
}
|
||||
]
|
||||
},
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
# Runtime Objects — Interstitial Contextual Messaging (P1-07, #30)
|
||||
|
||||
## Summary
|
||||
|
||||
Runtime objects are the observable plan/action/result state of a Rune turn,
|
||||
carried as typed, stable-id data on the operator's response envelope and
|
||||
projected into the TUI's **work pane**. They are the same events the chat log
|
||||
already narrates; the work pane is the operational index, the log remains the
|
||||
narrative record (**two-view principle**).
|
||||
|
||||
No chain-of-thought ever rides a runtime object: only what was actually
|
||||
admitted, run, observed, verified, or concluded.
|
||||
|
||||
## Object set (`src/rune/tool_protocol.py`)
|
||||
|
||||
All objects are frozen dataclasses with `to_dict()` / `from_dict()` round-trips
|
||||
and ISO-UTC `ts`. The pre-existing `ToolRequest` / `ToolEvent` got the same
|
||||
round-trip surface and are part of the set.
|
||||
|
||||
| Object | id field | Meaning |
|
||||
| --- | --- | --- |
|
||||
| `MemoryQuery` | `query_id` | One context-broker memory-selection pass: `admitted` (`{block_name: tokens}`), `dropped` (`[{name, tokens}]`), `memory_tokens`. |
|
||||
| `Observation` | `observation_id` | Observable runtime event: `kind` (`tool_result` / `approval` / `compaction` / `degraded` / `note`), one-line `summary`, bounded `detail` (<=300 chars). |
|
||||
| `VerificationResult` | `verification_id` | Objective verification of a task outcome: `method` (`exit_code` / `test` / `diff` / `manual` / `none`), `passed`, bounded `evidence`. Never self-report. |
|
||||
| `Outcome` | `outcome_id` | Final classified task outcome: `status` (`completed` / `failed` / `superseded` / `cancelled`), `verification_id` link ("" when none), `degraded`. |
|
||||
|
||||
## Emission points (`src/rune/rune_operator.py`)
|
||||
|
||||
Inside `_execute_on_model`'s tier loop, on the final response path (before
|
||||
`_finalize_runtime_response`):
|
||||
|
||||
1. **MemoryQuery** — one per broker `assemble`, built from `_last_trace`
|
||||
(admitted blocks → `{name: tokens}`, non-admitted blocks → dropped).
|
||||
2. **Observations** — one per broker compaction event (`kind=compaction`,
|
||||
summary carries the `tokens_before -> tokens_after` delta) and one per
|
||||
degraded subsystem name (`kind=degraded`, summary
|
||||
`subsystem unavailable: <name>`).
|
||||
3. **VerificationResult + Outcome** — emitted when the turn finishes without a
|
||||
pending approval. Method rule: `exit_code` when a `run_command` tool event
|
||||
completed successfully, `test` when `run_tests` completed, otherwise no
|
||||
VerificationResult is emitted at all (`method=none` is never claimed).
|
||||
`passed` = final response success AND no failed tool event. The Outcome's
|
||||
status is back-filled from the final task transition in `send_task`
|
||||
(`completed` / `failed`), and the verification link is set iff the
|
||||
verification passed **and** the task completed.
|
||||
|
||||
Emission is degrade-not-crash: a failure to build runtime objects records a
|
||||
`runtime_objects` subsystem failure (tier 2) and never fails the turn.
|
||||
|
||||
The objects flow to the TUI as dicts via the additive `runtime_objects` field
|
||||
on `OperatorResponse` (`_response_envelope` uses `asdict`, so the key survives
|
||||
automatically; default `[]` keeps pre-broker failure paths clean).
|
||||
|
||||
## Work pane (`src/rune/rune_tui.py`)
|
||||
|
||||
`WorkPane(Vertical)` is the third column in `#channels` (`id="work-pane"`,
|
||||
collapsible via Ctrl+3). CSS: `.work-pane { width: 1fr; border-left: solid
|
||||
$secondary; }`.
|
||||
|
||||
- **Data methods are headless-testable**: `add_object(obj)` / `update_object(obj)`
|
||||
(in-place by id, no duplicates) / `object_count()`; `render()` builds the
|
||||
display. Tests exercise the unmounted instance.
|
||||
- Sections render newest-first: Outcomes, Verification, Observations, Memory
|
||||
queries, Tool calls. Each line: kind glyph + stable id + one-line summary
|
||||
(+ bounded detail).
|
||||
- Tool calls keep their existing `ToolCallWidget` in the chat log AND get a
|
||||
one-line mirror in the pane (the pane is the operational index; the log
|
||||
keeps the interaction).
|
||||
- Visibility policy: hidden on the chat channel, visible on task; Ctrl+3
|
||||
toggles the expansion. `_consume_runtime_objects` feeds the pane from every
|
||||
response envelope (including approval resumes), degrade-not-crash.
|
||||
|
||||
## #33 handoff
|
||||
|
||||
`Outcome` is the OutcomeLedger's future record shape. This stage ships the
|
||||
typed object and its rendering only; ledger wiring (persisting outcomes) is
|
||||
the #33 dependency that adopts this shape. `VerificationResult` is the
|
||||
Verifier strategy's typed claim; the verifier itself is also #33 — until then
|
||||
the operator derives it from observable tool results (never self-report).
|
||||
@@ -0,0 +1,324 @@
|
||||
#!/usr/bin/env python3
|
||||
"""P1-07 runtime-objects canary (#30).
|
||||
|
||||
Self-bootstrapping like canary_p0_07.py (CI runs ``python3 <canary>``
|
||||
without PYTHONPATH). Proves the #30 claims are machine-checked with NO
|
||||
live model, NO Ollama, NO network:
|
||||
|
||||
1. all five runtime objects (ToolRequest/ToolEvent + MemoryQuery,
|
||||
Observation, VerificationResult, Outcome) round-trip to_dict/from_dict
|
||||
with stable ids;
|
||||
2. a scripted operator run (real RuneOperator, scripted broker/transport)
|
||||
emits a MemoryQuery matching the broker trace, a VerificationResult
|
||||
(method=exit_code, passed=True) and a linked Outcome (completed) on
|
||||
the response envelope;
|
||||
3. the WorkPane's headless data methods add/update objects by id, render
|
||||
kind glyphs + ids + summaries, and never duplicate.
|
||||
|
||||
Isolates all state under RUNE_HOME so the canary never touches a real
|
||||
home directory. Prints 'P1-07 RUNTIME OBJECTS CANARY PASSED' and exits 0.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
# Self-bootstrap: CI runs `python3 <canary>` without PYTHONPATH, so put the
|
||||
# repository src/ on sys.path before importing rune modules. Isolate Rune's
|
||||
# state/journal paths so the canary never touches a real home directory.
|
||||
_repo_src = str(Path(__file__).resolve().parents[1])
|
||||
sys.path.insert(0, _repo_src)
|
||||
_canary_home = tempfile.mkdtemp(prefix="p1-07-canary-home-")
|
||||
os.environ["RUNE_HOME"] = _canary_home
|
||||
os.environ["RUNE_MEMORY_DIR"] = tempfile.mkdtemp(prefix="p1-07-canary-mem-")
|
||||
|
||||
from rune import rune_operator, rune_tui # noqa: E402
|
||||
from rune.local_tool_runtime import OllamaChatTransport # noqa: E402
|
||||
from rune.tool_protocol import ( # noqa: E402
|
||||
MemoryQuery,
|
||||
Observation,
|
||||
OperatorResponse,
|
||||
Outcome,
|
||||
ToolEvent,
|
||||
ToolRequest,
|
||||
VerificationResult,
|
||||
)
|
||||
|
||||
TS = "2026-09-05T00:00:00+00:00"
|
||||
|
||||
|
||||
def _fail(message: str) -> int:
|
||||
print(f"FAILED: {message}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
class _ScriptedTransport(OllamaChatTransport):
|
||||
"""OllamaChatTransport with _post scripted — real accumulation code runs."""
|
||||
|
||||
def __init__(self, responses):
|
||||
super().__init__()
|
||||
self._responses = list(responses)
|
||||
|
||||
def _post(self, path, payload):
|
||||
if not self._responses:
|
||||
raise AssertionError("scripted transport exhausted")
|
||||
return self._responses.pop(0)
|
||||
|
||||
|
||||
def _chat_response(prompt_eval_count, eval_count, total_duration_ns):
|
||||
return {
|
||||
"model": "qwen3:8b",
|
||||
"created_at": "2026-09-05T00:00:00Z",
|
||||
"message": {"role": "assistant", "content": "ok"},
|
||||
"done": True,
|
||||
"prompt_eval_count": prompt_eval_count,
|
||||
"eval_count": eval_count,
|
||||
"total_duration": total_duration_ns,
|
||||
}
|
||||
|
||||
|
||||
class _ScriptedRuntime:
|
||||
"""Minimal runtime stand-in: one chat, then the canned final response."""
|
||||
|
||||
def __init__(self, transport, response):
|
||||
self.transport = transport
|
||||
self.tool_names = ("read_file", "write_file")
|
||||
self._response = response
|
||||
|
||||
def run(self, messages, **kwargs):
|
||||
self.transport.chat(model="qwen3:8b", messages=list(messages), tools=None)
|
||||
return self._response
|
||||
|
||||
|
||||
def _broker_trace(**overrides):
|
||||
trace = {
|
||||
"model_id": "qwen3:8b",
|
||||
"window": 40960,
|
||||
"budget": 32768,
|
||||
"blocks": [
|
||||
{"name": "persona", "tokens": 980, "admitted": True, "reason": ""},
|
||||
{"name": "tools", "tokens": 90, "admitted": True, "reason": ""},
|
||||
{"name": "memory", "tokens": 75, "admitted": True, "reason": ""},
|
||||
{"name": "history", "tokens": 400, "admitted": True, "reason": ""},
|
||||
{"name": "user", "tokens": 20, "admitted": True, "reason": ""},
|
||||
{
|
||||
"name": "consolidated",
|
||||
"tokens": 5000,
|
||||
"admitted": False,
|
||||
"reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
|
||||
},
|
||||
],
|
||||
"admitted": ["persona", "tools", "memory", "history", "user"],
|
||||
"admitted_total": 1565,
|
||||
"dropped": [
|
||||
{
|
||||
"name": "consolidated",
|
||||
"tokens": 5000,
|
||||
"reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
|
||||
}
|
||||
],
|
||||
"memory_tokens": 75,
|
||||
"degraded": ["memory_selector"],
|
||||
"compaction": [],
|
||||
"elapsed_ms": 1.5,
|
||||
}
|
||||
trace.update(overrides)
|
||||
return trace
|
||||
|
||||
|
||||
def check_round_trip() -> int:
|
||||
request = ToolRequest(call_id="call-1", name="read_file", arguments={"path": "/tmp/x"}, source="ollama-native")
|
||||
event = ToolEvent(kind="completed", request=request, result={"success": True})
|
||||
query = MemoryQuery(
|
||||
query_id="mq-1",
|
||||
task_id="task-1",
|
||||
admitted={"memory": 75},
|
||||
dropped=[{"name": "consolidated", "tokens": 5000}],
|
||||
memory_tokens=75,
|
||||
ts=TS,
|
||||
)
|
||||
observation = Observation(
|
||||
observation_id="obs-1",
|
||||
task_id="task-1",
|
||||
kind="degraded",
|
||||
summary="subsystem unavailable: memory_selector",
|
||||
detail="",
|
||||
ts=TS,
|
||||
)
|
||||
verification = VerificationResult(
|
||||
verification_id="vr-1",
|
||||
task_id="task-1",
|
||||
method="exit_code",
|
||||
passed=True,
|
||||
evidence="run_command exit 0",
|
||||
ts=TS,
|
||||
)
|
||||
outcome = Outcome(
|
||||
outcome_id="oc-1",
|
||||
task_id="task-1",
|
||||
status="completed",
|
||||
verification_id="vr-1",
|
||||
degraded=["memory_selector"],
|
||||
ts=TS,
|
||||
)
|
||||
for obj in (request, event, query, observation, verification, outcome):
|
||||
if type(obj).from_dict(obj.to_dict()) != obj:
|
||||
return _fail(f"{type(obj).__name__} did not round-trip")
|
||||
if MemoryQuery.from_dict(query.to_dict()).query_id != "mq-1":
|
||||
return _fail("MemoryQuery id not stable across round-trip")
|
||||
if Outcome.from_dict(outcome.to_dict()).verification_id != "vr-1":
|
||||
return _fail("Outcome verification link not stable")
|
||||
return 0
|
||||
|
||||
|
||||
def _scripted_operator_run():
|
||||
"""Real RuneOperator + scripted broker/transport → send_task envelope."""
|
||||
storage = Path(_canary_home) / "operator.json"
|
||||
op = rune_operator.RuneOperator(str(storage))
|
||||
trace = _broker_trace()
|
||||
capsule = SimpleNamespace(
|
||||
token_count=trace["admitted_total"],
|
||||
active_objective="",
|
||||
blocks=[],
|
||||
to_messages=lambda: [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "do it"},
|
||||
],
|
||||
)
|
||||
broker = SimpleNamespace(assemble=lambda task, session, model_id: (capsule, trace))
|
||||
profile = SimpleNamespace(safety_tier=0)
|
||||
response = OperatorResponse(
|
||||
success=True,
|
||||
user_text="Command ran clean.",
|
||||
model="qwen3:8b",
|
||||
tool_events=[
|
||||
ToolEvent(
|
||||
kind="completed",
|
||||
request=ToolRequest(call_id="call-1", name="run_command", arguments={}, source="ollama-native"),
|
||||
result={"success": True, "result": "(no output)"},
|
||||
)
|
||||
],
|
||||
)
|
||||
runtime = _ScriptedRuntime(_ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)]), response)
|
||||
patches = [
|
||||
patch.object(op, "_context_broker", return_value=broker),
|
||||
patch.object(rune_operator.model_profiles, "get_or_create_profile", return_value=profile),
|
||||
patch.object(rune_operator.model_profiles, "record_interaction"),
|
||||
patch.object(rune_operator, "LocalToolRuntime", return_value=runtime),
|
||||
patch.object(rune_operator, "OllamaChatTransport", return_value=runtime.transport),
|
||||
patch("rune.local_tool_runtime.context_length", lambda model: 40960),
|
||||
patch.object(
|
||||
rune_operator.safety_override.SafetyOverride,
|
||||
"get_params",
|
||||
lambda self, text: {"tier": 0, "temperature": 0.7, "prefix": "", "suffix": ""},
|
||||
),
|
||||
patch.object(rune_operator.safety_override.SafetyOverride, "is_refusal", lambda self, text: False),
|
||||
patch.object(op, "_log_token_usage"),
|
||||
]
|
||||
with contextlib.ExitStack() as stack:
|
||||
for item in patches:
|
||||
stack.enter_context(item)
|
||||
stack.enter_context(patch("rune.event_bus._store_path", lambda: Path(storage).parent / "journal-events.jsonl"))
|
||||
return op.send_task("inspect this", "qwen3:8b")
|
||||
|
||||
|
||||
def check_operator_emission() -> int:
|
||||
result = _scripted_operator_run()
|
||||
if not result.get("success"):
|
||||
return _fail("scripted turn did not succeed")
|
||||
objects = result.get("runtime_objects", [])
|
||||
if not isinstance(objects, list) or not objects:
|
||||
return _fail("runtime_objects missing from the response envelope")
|
||||
if not all(isinstance(obj, dict) for obj in objects):
|
||||
return _fail("runtime_objects must be dicts on the envelope")
|
||||
queries = [obj for obj in objects if "query_id" in obj]
|
||||
if len(queries) != 1:
|
||||
return _fail(f"expected exactly one MemoryQuery, got {len(queries)}")
|
||||
query = MemoryQuery.from_dict(queries[0])
|
||||
if query.admitted != {"persona": 980, "tools": 90, "memory": 75, "history": 400, "user": 20}:
|
||||
return _fail("MemoryQuery admitted blocks do not match the broker trace")
|
||||
if query.dropped != [{"name": "consolidated", "tokens": 5000}]:
|
||||
return _fail("MemoryQuery dropped blocks do not match the broker trace")
|
||||
if query.memory_tokens != 75:
|
||||
return _fail("MemoryQuery memory_tokens != 75")
|
||||
verifications = [obj for obj in objects if "method" in obj]
|
||||
if len(verifications) != 1:
|
||||
return _fail(f"expected exactly one VerificationResult, got {len(verifications)}")
|
||||
verification = VerificationResult.from_dict(verifications[0])
|
||||
if verification.method != "exit_code" or verification.passed is not True:
|
||||
return _fail("VerificationResult must be exit_code/passed after run_command exit 0")
|
||||
outcomes = [obj for obj in objects if "status" in obj]
|
||||
if len(outcomes) != 1:
|
||||
return _fail(f"expected exactly one Outcome, got {len(outcomes)}")
|
||||
outcome = Outcome.from_dict(outcomes[0])
|
||||
if outcome.status != "completed":
|
||||
return _fail(f"Outcome status must be completed, got {outcome.status}")
|
||||
if outcome.verification_id != verification.verification_id:
|
||||
return _fail("Outcome does not link the VerificationResult")
|
||||
return 0
|
||||
|
||||
|
||||
def check_work_pane() -> int:
|
||||
pane = rune_tui.WorkPane()
|
||||
observation = Observation(
|
||||
observation_id="obs-1",
|
||||
task_id="task-1",
|
||||
kind="degraded",
|
||||
summary="subsystem unavailable: memory_selector",
|
||||
detail="",
|
||||
ts=TS,
|
||||
).to_dict()
|
||||
outcome = Outcome(
|
||||
outcome_id="oc-1",
|
||||
task_id="task-1",
|
||||
status="completed",
|
||||
verification_id="",
|
||||
degraded=[],
|
||||
ts=TS,
|
||||
).to_dict()
|
||||
if not pane.add_object(observation) or not pane.add_object(outcome):
|
||||
return _fail("add_object rejected a new object")
|
||||
if pane.object_count() != 2:
|
||||
return _fail(f"object_count() = {pane.object_count()}, expected 2")
|
||||
rendered = pane.render().plain
|
||||
if "obs-1" not in rendered or "subsystem unavailable: memory_selector" not in rendered:
|
||||
return _fail("Observation glyph/id/summary missing from render")
|
||||
if "oc-1" not in rendered or "completed" not in rendered:
|
||||
return _fail("Outcome glyph/id/status missing from render")
|
||||
if rendered.index("oc-1") >= rendered.index("obs-1"):
|
||||
return _fail("sections must render newest-first (Outcomes before Observations)")
|
||||
updated = {**observation, "summary": "subsystem unavailable: storage"}
|
||||
if not pane.update_object(updated):
|
||||
return _fail("update_object rejected a known id")
|
||||
if pane.object_count() != 2:
|
||||
return _fail("update_object duplicated an entry")
|
||||
if pane.add_object(updated):
|
||||
return _fail("re-adding a known id must be an in-place update (False)")
|
||||
if pane.object_count() != 2:
|
||||
return _fail("re-adding a known id duplicated an entry")
|
||||
if "memory_selector" in pane.render().plain or "storage" not in pane.render().plain:
|
||||
return _fail("update_object did not replace the summary in place")
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
checks = (
|
||||
("all five runtime objects round-trip with stable ids", check_round_trip),
|
||||
("scripted operator emits MemoryQuery + VerificationResult + Outcome", check_operator_emission),
|
||||
("work pane data methods render/update by id", check_work_pane),
|
||||
)
|
||||
for label, check in checks:
|
||||
result = check()
|
||||
if result != 0:
|
||||
return result
|
||||
print(f" ✓ {label}")
|
||||
print("P1-07 RUNTIME OBJECTS CANARY PASSED")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,511 @@
|
||||
#!/usr/bin/env python3
|
||||
"""memory_selector.py — the ONE memory-selection API (P1-03, #26).
|
||||
|
||||
Authority map (config/v1_authority_map.json, domain ``memory_selection``):
|
||||
``ContextBroker.memory_selector`` is the v1_owner. Everything that surfaces
|
||||
memory into model context flows through ``MemorySelector.select_for_task`` —
|
||||
the operator registers exactly ONE memory contributor consuming it; the
|
||||
pre-P1-03 parallel contributors (recall, persona_memories, consolidated,
|
||||
neurons, graphs, abstraction) are replaced, not added to.
|
||||
|
||||
Policy (issue #26):
|
||||
- ONE relevance floor — ``memory_store.RECALL_MIN_SCORE`` (0.42) default —
|
||||
applied to EVERY source. The store source applies it inside the store's
|
||||
calibrated gate (``min_score`` is passed down into ``recall``, which
|
||||
enforces it in EVERY mode: lexical-only, dense-only, hybrid); self-
|
||||
scored sources are floored here. ``gate_consolidated``'s flat-0.8
|
||||
shortcut inside ``cached_recall`` is removed (Task 3) so consolidated
|
||||
knowledge competes under the same floor, and the top-5-importance neuron
|
||||
fallback is gone.
|
||||
- TRUE ABSTENTION: a trivial task legitimately receives ZERO recalled
|
||||
memories. Below-floor candidates are counted (the irrelevant-memory rate
|
||||
is computable per run) but never selected.
|
||||
- Memory-type targeting via ``memory_store``'s typed recall filter.
|
||||
- FAIL-CLOSED: a source failure is a skip + log, not an injection; if the
|
||||
selector itself is unavailable the contributor abstains — never fail open
|
||||
to "inject something important".
|
||||
|
||||
Degrade-not-crash throughout: sources are pluggable and isolated; one
|
||||
exploding source never takes down assembly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import inspect
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable
|
||||
|
||||
try:
|
||||
from . import memory_store
|
||||
except ImportError: # direct script execution
|
||||
import memory_store # type: ignore[no-redef]
|
||||
|
||||
try:
|
||||
from .context_broker import ContributionBlock
|
||||
except ImportError: # direct script execution
|
||||
from context_broker import ContributionBlock # type: ignore[no-redef]
|
||||
|
||||
LOG = logging.getLogger("rune.memory_selector")
|
||||
|
||||
# Candidate cap BEFORE budgeting (plan DD 5: the selector caps its own
|
||||
# candidate list before returning).
|
||||
MAX_CANDIDATES = 24
|
||||
|
||||
# Default per-task candidate budget (the old memory contributor recalled 3;
|
||||
# the collapsed selector merges all sources under one slightly larger cap).
|
||||
DEFAULT_BUDGET = 4
|
||||
|
||||
# Priority of the single memory block (was _contributor_memory's 70).
|
||||
MEMORY_BLOCK_PRIORITY = 70
|
||||
|
||||
|
||||
@dataclass
|
||||
class SelectionResult:
|
||||
"""One selection outcome — mirror of CapabilityBroker's select_for_job.
|
||||
|
||||
``irrelevant_rate`` is computable per run:
|
||||
``(candidates_evaluated - selected_count) / candidates_evaluated`` when
|
||||
candidates > 0; abstained runs are recorded as zero-injection events.
|
||||
"""
|
||||
|
||||
blocks: list[ContributionBlock]
|
||||
candidates_evaluated: int
|
||||
selected_count: int
|
||||
abstained: bool
|
||||
types_queried: list[str] = field(default_factory=list)
|
||||
|
||||
def irrelevant_rate(self) -> float:
|
||||
if self.candidates_evaluated <= 0:
|
||||
return 0.0
|
||||
return (self.candidates_evaluated - self.selected_count) / self.candidates_evaluated
|
||||
|
||||
|
||||
class MemorySelector:
|
||||
"""Single memory-selection API composed by ContextBroker.
|
||||
|
||||
Sources are registered callables ``(query, types, budget, min_score) ->
|
||||
list[dict]`` where each dict carries at least ``text`` and ``score``
|
||||
(0-1 relevance under the ONE floor). The store source (``_store_recall``)
|
||||
routes through ``memory_store.recall``'s calibrated gate; Task 3 adds the
|
||||
remaining sources. A source failure is skipped + logged (fail-closed).
|
||||
|
||||
KNOWN RANKING BIAS (P1-03 #26 discharge, finding 9): the ONE floor gates
|
||||
each source on its OWN calibrated scale (lexical store ~4-6, dense/
|
||||
hybrid store RRF ~0-0.13 with the cosine gated at 0.42, self-scored
|
||||
0-1), but merged candidates are RANKED by raw score — so a lexical-mode
|
||||
store row outranks an equally-relevant self-scored row. Abstention is
|
||||
unaffected (every source is floored on its own scale); only candidate
|
||||
ORDER is biased. Per-source rank normalization was deliberately deferred:
|
||||
renormalizing the store's calibrated scales risked false abstentions
|
||||
this late in the release.
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: str | None = None, min_score: float | None = None):
|
||||
self.db_path = db_path if db_path is not None else memory_store.DB_PATH
|
||||
self.min_score = float(min_score) if min_score is not None else float(memory_store.RECALL_MIN_SCORE)
|
||||
# P1-03 (#26) DISCHARGE (finding 3): the selector's PINNED db_path is
|
||||
# bound into store-backed sources at registration time — a bare
|
||||
# ``memory_store.connect()`` inside a source used to open the
|
||||
# module-default store, ignoring the pin. Only sources that actually
|
||||
# declare a ``db_path`` parameter are bound (replacement sources
|
||||
# registered via add_source keep the documented 4-arg contract).
|
||||
self._sources: dict[str, Callable] = {}
|
||||
for name, fn in DEFAULT_SOURCES.items():
|
||||
try:
|
||||
params = inspect.signature(fn).parameters
|
||||
has_db_path = "db_path" in params or any(
|
||||
p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
has_db_path = False
|
||||
if has_db_path:
|
||||
self._sources[name] = functools.partial(fn, db_path=self.db_path)
|
||||
else:
|
||||
self._sources[name] = fn
|
||||
|
||||
# -- source registry -----------------------------------------------------
|
||||
|
||||
def add_source(self, name: str, fn: Callable) -> None:
|
||||
"""Register a selection source. Registration ORDER is execution order
|
||||
(dict preserves it); re-registering a name replaces that source."""
|
||||
self._sources[name] = fn
|
||||
|
||||
def source_names(self) -> list[str]:
|
||||
"""Registered source names in execution order."""
|
||||
return list(self._sources)
|
||||
|
||||
# -- selection -------------------------------------------------------------
|
||||
|
||||
def select_for_task(
|
||||
self,
|
||||
task_description: str,
|
||||
types: list[str] | None = None,
|
||||
budget: int = DEFAULT_BUDGET,
|
||||
) -> SelectionResult:
|
||||
"""Select memory for one task under the ONE floor.
|
||||
|
||||
Returns a SelectionResult; on total selector failure the caller
|
||||
(operator contributor) treats any exception as an abstention.
|
||||
"""
|
||||
query = (task_description or "").strip()
|
||||
types_queried = list(types) if types else []
|
||||
if not query:
|
||||
return SelectionResult([], 0, 0, True, types_queried)
|
||||
|
||||
candidates: list[dict] = []
|
||||
for name, fn in self._sources.items():
|
||||
try:
|
||||
found = fn(query, types_queried, MAX_CANDIDATES, self.min_score)
|
||||
except Exception as exc: # fail-closed per source
|
||||
LOG.warning("memory source %r failed; skipping (%s)", name, exc)
|
||||
continue
|
||||
for item in found or []:
|
||||
entry = dict(item)
|
||||
entry["kind"] = entry.get("kind") or name
|
||||
candidates.append(entry)
|
||||
candidates = candidates[:MAX_CANDIDATES]
|
||||
|
||||
# ONE floor for EVERY source: the store source was already floored by
|
||||
# the same constant inside recall() — which applies min_score in
|
||||
# EVERY mode (lexical-only, dense-only, hybrid; P1-03 #26 discharge)
|
||||
# — so its rows are tagged store_scored and exempt from this
|
||||
# selector-side numeric check. Self-scored sources are floored here
|
||||
# against the same selector min_score.
|
||||
selected = [item for item in candidates if item.get("store_scored") or _score_of(item) >= self.min_score]
|
||||
selected.sort(key=lambda item: -_score_of(item))
|
||||
selected = selected[: max(0, budget)]
|
||||
|
||||
blocks: list[ContributionBlock] = []
|
||||
for item in selected:
|
||||
text = str(item.get("text", "")).strip()
|
||||
if not text:
|
||||
continue
|
||||
blocks.append(
|
||||
ContributionBlock(
|
||||
"memory",
|
||||
MEMORY_BLOCK_PRIORITY,
|
||||
f"- {text}",
|
||||
max(1, len(text) // 4),
|
||||
"memory_selector.select_for_task",
|
||||
evictable=True,
|
||||
)
|
||||
)
|
||||
|
||||
abstained = not blocks
|
||||
if abstained:
|
||||
LOG.info(
|
||||
"memory abstain: task=%r candidates=%d floor=%.2f types=%s",
|
||||
query[:60],
|
||||
len(candidates),
|
||||
self.min_score,
|
||||
types_queried or "all",
|
||||
)
|
||||
return SelectionResult(blocks, len(candidates), len(blocks), abstained, types_queried)
|
||||
|
||||
# -- sources ---------------------------------------------------------------
|
||||
|
||||
def _source_store(self, query: str, types: list[str], budget: int, min_score: float) -> list[dict]:
|
||||
"""Typed store recall through the ONE thresholded path."""
|
||||
return _store_recall(query, types, budget, min_score, db_path=self.db_path)
|
||||
|
||||
|
||||
def _score_of(item: dict) -> float:
|
||||
"""Extract the item's 0-1 relevance under the ONE floor. Missing or
|
||||
malformed scores floor at 0.0 (abstain) — never guessed upward."""
|
||||
value = item.get("score")
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
return 0.0
|
||||
return float(value)
|
||||
|
||||
|
||||
def _store_recall(
|
||||
query: str,
|
||||
types: list[str] | None,
|
||||
budget: int,
|
||||
min_score: float = memory_store.RECALL_MIN_SCORE,
|
||||
db_path: str | None = None,
|
||||
) -> list[dict]:
|
||||
"""Store source: typed recall through the ONE thresholded path.
|
||||
|
||||
Uses ``memory_store.recall`` (the calibrated QPP gate) — never the
|
||||
``cached_recall`` path whose ``gate_consolidated`` flat-0.8 shortcut
|
||||
bypasses thresholding (removed in Task 3).
|
||||
|
||||
The store's ``relevance`` lives on mode-dependent calibrated scales
|
||||
(lexical-mode ~4-6 via the +5 offset band, dense/hybrid-mode RRF ~0-0.13,
|
||||
with the dense COSINE gated at 0.42-0.8 — see memory_store.recall), so it
|
||||
is NOT renormalized here: the store already applied THE floor via
|
||||
``min_score`` in EVERY recall mode (lexical-only, dense-only, hybrid —
|
||||
P1-03 #26 discharge), so tagging the result ``store_scored=True`` and
|
||||
exempting it from the selector's numeric check is sound. The result is
|
||||
tagged ``store_scored=True`` so the selector treats these rows as
|
||||
floor-cleared by the same constant.
|
||||
|
||||
KNOWN RANKING BIAS (P1-03 #26 discharge, finding 9): because the scales
|
||||
differ per mode, the selector's raw-score ranking favors lexical-mode
|
||||
store rows (~4-6) over self-scored rows (0-1) and dense/hybrid-mode
|
||||
store rows (~0-0.13). Abstention is NOT affected (each source is floored
|
||||
on its own calibrated scale); only the ORDER of merged candidates is
|
||||
biased. Per-source rank normalization was deliberately deferred —
|
||||
renormalizing the store's calibrated scales risked false abstentions.
|
||||
|
||||
P1-03 (#26) DISCHARGE (finding 2a): besides the caller-typed query, the
|
||||
source always runs a SECOND typed query restricted to
|
||||
``self``/``personality`` — the persona lane. The production caller (the
|
||||
operator's single memory contributor) invokes the selector WITHOUT
|
||||
types, which left the typed persona source dead in production. Both
|
||||
queries are thresholded inside the store's gate and may abstain — a
|
||||
trivial task still gets zero persona rows. Results are merged and
|
||||
capped at ``budget``.
|
||||
|
||||
P1-03 (#26) DEVIATION: shard routing inside ``recall`` is bypassed — the
|
||||
selector pins the canonical store connection. ShardManager routes to a
|
||||
hardcoded ~/.config/opencode shard dir that ignores RUNE_* isolation and
|
||||
carries no memory_type column, so routing through it would break typed
|
||||
recall and the preserve-stores/isolation invariants.
|
||||
"""
|
||||
db = memory_store.connect(db_path)
|
||||
try:
|
||||
results = memory_store.recall(db, query, budget=budget, min_score=min_score, memory_types=types or None)
|
||||
persona_types = ["self", "personality"]
|
||||
if not (set(types or []) & set(persona_types)):
|
||||
# The persona lane: always executed for the untyped production
|
||||
# caller (and any caller that did not itself ask persona types).
|
||||
results = results + memory_store.recall(
|
||||
db, query, budget=budget, min_score=min_score, memory_types=persona_types
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
out: list[dict] = []
|
||||
seen_ids: set = set()
|
||||
for row in results:
|
||||
# The untyped and typed-lane queries overlap on unclassified rows —
|
||||
# merge by entry id so a shared row is a candidate ONCE (duplicates
|
||||
# would corrupt the irrelevant-memory rate and repeat block text).
|
||||
row_id = row.get("id")
|
||||
if row_id is not None:
|
||||
if row_id in seen_ids:
|
||||
continue
|
||||
seen_ids.add(row_id)
|
||||
out.append(
|
||||
{
|
||||
"text": row.get("text", ""),
|
||||
"score": float(row.get("relevance", 0.0)),
|
||||
"store_scored": True,
|
||||
"kind": "store",
|
||||
"id": row.get("id"),
|
||||
"source": row.get("source", ""),
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _lex_overlap_score(text: str, query: str) -> float:
|
||||
"""Deterministic lexical relevance for self-scored sources.
|
||||
|
||||
Distinctive-word coverage of the QUERY terms present in the candidate
|
||||
text, with a stopword filter shared with the store's gate vocabulary.
|
||||
A candidate sharing NO distinctive query word cannot score — the same
|
||||
conjunctive logic the store's BM25 gate applies (a shared single word is
|
||||
not a relation, but zero shared words is definitely not one).
|
||||
"""
|
||||
stopwords = getattr(memory_store, "_STOPWORDS", set())
|
||||
q_terms = {t.lower() for t in query.split() if len(t) > 3 and t.lower() not in stopwords}
|
||||
if not q_terms:
|
||||
return 0.0
|
||||
text_lower = (text or "").lower()
|
||||
hits = sum(1 for term in q_terms if term in text_lower)
|
||||
return hits / len(q_terms)
|
||||
|
||||
|
||||
def _neurons_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
|
||||
"""Warm-neuron source: trigger scan, thresholded — NO importance fallback.
|
||||
|
||||
The pre-P1-03 contributor injected the top-5 neurons by importance when
|
||||
no trigger matched (rune_operator.py:1387-1389). That fallback is dead:
|
||||
a neuron that does not match the task is not injected merely because it
|
||||
is important. Trigger-matched neurons are scored lexically against the
|
||||
query and must clear the same floor as every other source.
|
||||
"""
|
||||
try:
|
||||
import warm_neuron_store
|
||||
except ImportError: # TUI top-level deploy
|
||||
import rune.warm_neuron_store as warm_neuron_store # type: ignore[no-redef]
|
||||
|
||||
neurons = warm_neuron_store.list_neurons()
|
||||
task_words = {w.lower().strip(".,;:!?()[]{}\"'") for w in query.split()}
|
||||
out: list[dict] = []
|
||||
for neuron in neurons:
|
||||
triggers = neuron.get("triggers") or []
|
||||
if not any(t in task_words for t in triggers):
|
||||
continue
|
||||
text = neuron.get("text", "")
|
||||
score = _lex_overlap_score(text, query)
|
||||
if score < min_score:
|
||||
continue
|
||||
out.append(
|
||||
{
|
||||
"text": f"[{neuron.get('slug', 'warm')}] {text}",
|
||||
"score": score,
|
||||
"kind": "neuron",
|
||||
"source": "warm_neuron_store.list_neurons",
|
||||
}
|
||||
)
|
||||
if len(out) >= budget:
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def _graphs_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
|
||||
"""Graph source: entity BFS from the task words via graph_memory.query,
|
||||
floored. The pre-P1-03 contributor called resolve_entity(), which MUTATES
|
||||
the graph (creates missing entities); read-only query() keeps the store
|
||||
side-effect-free. A term with no entity yields no candidates."""
|
||||
try:
|
||||
import graph_memory
|
||||
except ImportError: # TUI top-level deploy
|
||||
import rune.graph_memory as graph_memory # type: ignore[no-redef]
|
||||
|
||||
db = graph_memory.connect()
|
||||
out: list[dict] = []
|
||||
try:
|
||||
task_words = list(dict.fromkeys(w.lower().strip(".,;:!?()[]{}\"'") for w in query.split()))[:5]
|
||||
for word in task_words:
|
||||
result = graph_memory.query(db, word, hops=1)
|
||||
if not result:
|
||||
continue
|
||||
entity = result.get("entity", word) if isinstance(result, dict) else str(result)
|
||||
facts = result.get("facts", []) if isinstance(result, dict) else []
|
||||
lines = [f"{f.get('subject')} {f.get('predicate', '-')} {f.get('object')}" for f in facts]
|
||||
text = "; ".join(lines) if lines else str(entity)
|
||||
score = _lex_overlap_score(text, query) or (1.0 / max(len(task_words), 1))
|
||||
if score < min_score:
|
||||
continue
|
||||
out.append({"text": text, "score": score, "kind": "graph", "source": "graph_memory.query"})
|
||||
if len(out) >= budget:
|
||||
break
|
||||
finally:
|
||||
db.close()
|
||||
return out
|
||||
|
||||
|
||||
def _abstraction_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
|
||||
"""Abstraction-facts source: LIKE scan over active facts, floored.
|
||||
|
||||
The pre-P1-03 contributor injected the top-3 LIKE matches unconditionally;
|
||||
here a LIKE match is only a candidate — it must clear the floor.
|
||||
"""
|
||||
try:
|
||||
import abstraction
|
||||
except ImportError: # TUI top-level deploy
|
||||
import rune.abstraction as abstraction # type: ignore[no-redef]
|
||||
|
||||
db = abstraction.connect()
|
||||
out: list[dict] = []
|
||||
try:
|
||||
task_words = [w for w in query.lower().split()[:3] if w]
|
||||
if not task_words:
|
||||
return []
|
||||
placeholders = " OR ".join(["what LIKE ?" for _ in task_words])
|
||||
params = [f"%{w}%" for w in task_words]
|
||||
rows = db.execute(
|
||||
f"SELECT what FROM facts WHERE state='active' AND ({placeholders}) LIMIT {int(budget) * 3}",
|
||||
params,
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
text = str(row[0])
|
||||
score = _lex_overlap_score(text, query)
|
||||
if score < min_score:
|
||||
continue
|
||||
out.append({"text": text, "score": score, "kind": "abstraction", "source": "abstraction facts"})
|
||||
if len(out) >= budget:
|
||||
break
|
||||
finally:
|
||||
db.close()
|
||||
return out
|
||||
|
||||
|
||||
def _consolidated_recall(query: str, types: list[str] | None, budget: int, min_score: float) -> list[dict]:
|
||||
"""Consolidated-knowledge source under the SAME floor (bypass removed).
|
||||
|
||||
The pre-P1-03 path handed ``gate_consolidated``'s output a flat
|
||||
relevance=0.8, bypassing thresholding entirely. Consolidated knowledge
|
||||
now competes like any candidate: EACH knowledge item is scored by its
|
||||
lexical overlap with the active task, and below-floor items abstain.
|
||||
"""
|
||||
try:
|
||||
from .memory_consolidation import MemoryConsolidation
|
||||
except ImportError: # TUI top-level deploy
|
||||
from memory_consolidation import MemoryConsolidation # type: ignore[no-redef]
|
||||
|
||||
try:
|
||||
consolidation = MemoryConsolidation()
|
||||
items = list(consolidation.get_relevant_knowledge(query, limit=budget * 2))
|
||||
except Exception as exc: # degrade: consolidation unavailable -> no candidates
|
||||
LOG.warning("consolidation source unavailable; skipping (%s)", exc)
|
||||
return []
|
||||
out: list[dict] = []
|
||||
for item in items:
|
||||
content = (getattr(item, "content", "") or "").strip()
|
||||
if not content:
|
||||
continue
|
||||
score = _lex_overlap_score(content, query)
|
||||
if score < min_score:
|
||||
continue # below floor -> abstain (was: flat relevance injection)
|
||||
out.append({"text": content, "score": score, "kind": "consolidated", "source": "memory_consolidation"})
|
||||
if len(out) >= budget:
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def _persona_recall(
|
||||
query: str,
|
||||
types: list[str] | None,
|
||||
budget: int,
|
||||
min_score: float,
|
||||
db_path: str | None = None,
|
||||
) -> list[dict]:
|
||||
"""Typed persona source: self/personality memory keyed on the ACTIVE task.
|
||||
|
||||
Replaces the hardcoded ``"Rune personality voice identity how to speak"``
|
||||
query. Only runs when the caller asked for self/personality types (the
|
||||
operator's persona-memory call site); thresholded, may abstain — a
|
||||
trivial task gets zero persona lines.
|
||||
|
||||
P1-03 (#26) DISCHARGE (finding 3): the source queries the selector's
|
||||
PINNED store (``db_path``) — never the module-default store that a bare
|
||||
``memory_store.connect()`` would open.
|
||||
"""
|
||||
wanted = {t for t in (types or []) if t in ("self", "personality")}
|
||||
if not wanted:
|
||||
return []
|
||||
db = memory_store.connect(db_path)
|
||||
try:
|
||||
rows = memory_store.recall(db, query, budget=budget, min_score=min_score, memory_types=sorted(wanted))
|
||||
finally:
|
||||
db.close()
|
||||
out: list[dict] = []
|
||||
for row in rows:
|
||||
out.append(
|
||||
{
|
||||
"text": row.get("text", ""),
|
||||
"score": float(row.get("relevance", 0.0)),
|
||||
"kind": "persona",
|
||||
"source": "memory_store.recall(self/personality)",
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
# The default source set (P1-03 #26): every pre-selector memory path, unified
|
||||
# under the ONE floor. Registration order = execution order.
|
||||
DEFAULT_SOURCES: dict[str, Callable] = {
|
||||
"store": _store_recall,
|
||||
"neurons": _neurons_recall,
|
||||
"graphs": _graphs_recall,
|
||||
"abstraction": _abstraction_recall,
|
||||
"consolidated": _consolidated_recall,
|
||||
"persona": _persona_recall,
|
||||
}
|
||||
@@ -470,7 +470,7 @@ def add_entry(db, text, importance=0.5, topic="", entities=None,
|
||||
return True
|
||||
|
||||
|
||||
def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE):
|
||||
def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE, *, probe_cap: int = RECALL_BUDGET * 2):
|
||||
"""Hybrid recall: dense + BM25 fused by RRF, metadata-aware.
|
||||
|
||||
Returns (entries, scores) — the top `budget` active entries. If nothing
|
||||
@@ -486,7 +486,17 @@ def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE):
|
||||
|
||||
CACHE LAYER (2026-08-21): check cache before expensive retrieval.
|
||||
Cache is invalidated by event bus when memories change.
|
||||
|
||||
M8 (#161): the internal dense/FTS probe caps are parameterized via
|
||||
`probe_cap` (default RECALL_BUDGET*2 — byte-identical behavior when
|
||||
unset). The benchmark harness sweeps this alongside top-k to expose
|
||||
candidate-pool starvation vs rank-ordering failures (INV-5).
|
||||
|
||||
M1 (#161): every returned entry carries `mode` — 'lexical-only',
|
||||
'dense-only', or 'hybrid' — recording which fusion branch produced it.
|
||||
Observability ONLY (INV-13): no consumer may branch behavior on it.
|
||||
"""
|
||||
|
||||
query = (query or "").strip()
|
||||
if not query:
|
||||
return []
|
||||
@@ -524,7 +534,7 @@ def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE):
|
||||
rows = db.execute(
|
||||
"SELECT v.rowid, distance FROM entries_vec v "
|
||||
"WHERE v.embedding MATCH ? AND k = ?",
|
||||
(json.dumps(qvec), RECALL_BUDGET * 2)).fetchall()
|
||||
(json.dumps(qvec), probe_cap)).fetchall()
|
||||
# META-JUDGEMENT (applies in BOTH modes): vector proximity is not a
|
||||
# relation by itself. A dense hit must also share a CONTENT WORD with
|
||||
# the query — otherwise "black coffee" would pull in "black holes"
|
||||
@@ -574,7 +584,7 @@ def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE):
|
||||
"SELECT rowid, bm25(entries_fts) AS score FROM entries_fts "
|
||||
"WHERE entries_fts MATCH ? "
|
||||
"ORDER BY bm25(entries_fts) LIMIT ?",
|
||||
(_fts_query_conjunctive(query), RECALL_BUDGET * 2)).fetchall()
|
||||
(_fts_query_conjunctive(query), probe_cap)).fetchall()
|
||||
# Keep REAL BM25 scores (Robertson & Zaragoza 2009) for the QPP
|
||||
# threshold — never flatten them.
|
||||
bm25 = [(r["rowid"], float(r["score"])) for r in rows]
|
||||
@@ -600,7 +610,7 @@ def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE):
|
||||
except Exception:
|
||||
pass
|
||||
if _expanded_hits:
|
||||
_ids = list(_expanded_hits)[:RECALL_BUDGET * 2]
|
||||
_ids = list(_expanded_hits)[:probe_cap]
|
||||
_ph = ",".join("?" * len(_ids))
|
||||
_rows = db.execute(
|
||||
f"SELECT rowid, bm25(entries_fts) AS score FROM entries_fts "
|
||||
@@ -614,13 +624,19 @@ def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE):
|
||||
# arXiv:2210.11934) — use raw BM25 scores directly.
|
||||
ranks = {}
|
||||
best_dense = 0.0
|
||||
# M1 observability (#161 synthesis INV-13): which fusion branch produced
|
||||
# the results. Observability ONLY — consumers must never branch behavior
|
||||
# on this field (that would create a second retrieval authority).
|
||||
retrieval_mode = "lexical-only"
|
||||
if dense and bm25:
|
||||
retrieval_mode = "hybrid"
|
||||
for idx, lst in enumerate([dense, bm25]):
|
||||
for pos, (eid, _score) in enumerate(lst):
|
||||
ranks[eid] = ranks.get(eid, 0.0) + 1.0 / (RRF_K + pos + 1)
|
||||
if idx == 0:
|
||||
best_dense = max(best_dense, _score)
|
||||
elif dense:
|
||||
retrieval_mode = "dense-only"
|
||||
for pos, (eid, _score) in enumerate(dense):
|
||||
ranks[eid] = 1.0 / (RRF_K + pos + 1)
|
||||
best_dense = max(best_dense, _score)
|
||||
@@ -687,6 +703,9 @@ def recall(db, query, budget=RECALL_BUDGET, min_score=RECALL_MIN_SCORE):
|
||||
"confidence": round(conf, 2),
|
||||
"temperature": round(temp, 2),
|
||||
"relevance": round(relevance, 4),
|
||||
# M1 observability (#161): lexical-only | dense-only | hybrid.
|
||||
# NEVER consumed for behavior — see INV-13 in the synthesis.
|
||||
"mode": retrieval_mode,
|
||||
})
|
||||
# Enforce budget + abstention floor.
|
||||
if not scored:
|
||||
|
||||
@@ -0,0 +1,330 @@
|
||||
#!/usr/bin/env python3
|
||||
"""retrieval_benchmark.py — retrieval-quality benchmark harness (#161).
|
||||
|
||||
Accepted synthesis e584bbf895b3 (research-161-r1, canonical RP7): build ONE
|
||||
harness that consumes memory_selector's EXISTING source registry as its only
|
||||
backend abstraction (M5/INV-8 — no new adapter protocol), reports backend-
|
||||
local precision-recall curves at stated operating points (M4/INV-10 — never
|
||||
a single thresholded cross-backend number), an abstention-aware headline
|
||||
metric (M7/INV-11), stage-by-stage latency profiles (M8), and neutralized-
|
||||
metadata quality rows alongside live-composite deltas (M2/INV-3).
|
||||
|
||||
Carried adversarial constraints (mandatory, not prose):
|
||||
M1/INV-13 mode field is observability-only; harness fails loudly when the
|
||||
embedding service is unreachable (FT1/INV-1: no silent mode
|
||||
switch — an unlabeled degradation voids validation)
|
||||
M2/INV-3 quality metrics computed with metadata neutralized
|
||||
(confidence=temperature=1.0, recency frozen) via a frozen
|
||||
reference clock; live-vs-neutral delta is first-class output
|
||||
M3 result cache and _embed cache cleared/bypassed between
|
||||
measured runs; warm-up warms the embedding server only
|
||||
M4/INV-10 backend-local PR curves + ranking-bias disclosure in every
|
||||
cross-source row
|
||||
M5/INV-8 backends are memory_selector registry sources — the existing
|
||||
(query, types, budget, min_score) contract; no second protocol
|
||||
M6/INV-9 benchmark stores created exclusively under RUNE_*-isolated
|
||||
scratch paths; DB_PATH != memory_env.store_db(); no shard dirs
|
||||
M8 stage profiler: embed / dense_gate / fts / rrf / metadata_blend
|
||||
/ abstention instrumented separately; probe_cap swept with top-k
|
||||
M9/INV-7 no E20-E22 numeric claim enters decision inputs without primary-
|
||||
source verification (evidence-citation gate)
|
||||
M10/INV-12 one shared run-record schema; fingerprint() reused verbatim
|
||||
from orchestration/benchmark.py
|
||||
|
||||
DO NOT consume the mode field for behavior. DO NOT add a second backend
|
||||
protocol. DO NOT report cross-backend numbers without operating points.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sqlite3
|
||||
import statistics
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from rune.orchestration.benchmark import fingerprint
|
||||
|
||||
|
||||
# ── M6/INV-9: store isolation ────────────────────────────────────────────────
|
||||
|
||||
class IsolationError(RuntimeError):
|
||||
"""Raised when a benchmark store path violates the isolation contract."""
|
||||
|
||||
|
||||
def assert_isolated_store(db_path: str) -> None:
|
||||
"""INV-9: every benchmark store must be RUNE_*-isolated. Checked BEFORE
|
||||
any row is written. Rejects the production store, shard dirs, and any
|
||||
path outside a RUNE_* scratch root."""
|
||||
resolved = os.path.abspath(db_path)
|
||||
parts = Path(resolved).parts
|
||||
if not any(p.startswith("RUNE_") for p in parts):
|
||||
raise IsolationError(
|
||||
f"benchmark store {resolved} is not under a RUNE_* scratch root")
|
||||
try:
|
||||
from rune import memory_env
|
||||
prod = os.path.abspath(memory_env.store_db())
|
||||
if os.path.abspath(resolved) == prod:
|
||||
raise IsolationError("benchmark store must not be memory_env.store_db()")
|
||||
except IsolationError:
|
||||
raise
|
||||
except Exception:
|
||||
pass # memory_env unavailable in test contexts — path check above stands
|
||||
if "/shards/" in resolved or resolved.endswith(".shard.db"):
|
||||
raise IsolationError("benchmark store must not touch shard directories")
|
||||
|
||||
|
||||
def make_isolated_store(entries: list[dict]) -> tuple[sqlite3.Connection, str]:
|
||||
"""Create a fully isolated benchmark store under a RUNE_* temp root.
|
||||
Uses the PRODUCTION schema via memory_store.connect() — the store's own
|
||||
DDL, not a harness-written copy (no second schema authority).
|
||||
Returns (db, db_path)."""
|
||||
root = tempfile.mkdtemp(prefix="RUNE_bench_")
|
||||
db_path = os.path.join(root, "bench.db")
|
||||
assert_isolated_store(db_path)
|
||||
from rune import memory_store as ms
|
||||
# Point the store module at the isolated path for this connection, then
|
||||
# restore. The store's connect() reads module-level DB_PATH.
|
||||
original_db_path = ms.DB_PATH
|
||||
ms.DB_PATH = db_path
|
||||
try:
|
||||
os.makedirs(root, exist_ok=True)
|
||||
db = ms.connect()
|
||||
finally:
|
||||
ms.DB_PATH = original_db_path
|
||||
for e in entries:
|
||||
ms.add_entry(db, e["text"], importance=e.get("importance", 0.5),
|
||||
topic=e.get("topic", ""), entities=e.get("entities", []),
|
||||
source=e.get("source", "benchmark"))
|
||||
return db, db_path
|
||||
|
||||
|
||||
# ── M3: cache bypass ─────────────────────────────────────────────────────────
|
||||
|
||||
def bypass_caches() -> dict:
|
||||
"""M3: clear/bypass recall()'s result cache and _embed cache between
|
||||
measured runs. Returns the bypass state for auditability."""
|
||||
cleared = {"result_cache": False, "embed_cache": False}
|
||||
try:
|
||||
import cache as _cache
|
||||
# The result cache is keyed "recall:<query>"; clear known keys is not
|
||||
# enumerable — instead the harness measures via direct recall() calls
|
||||
# with the cache module's get_cache patched out at the call site.
|
||||
cleared["result_cache"] = True
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
from rune import memory_store as ms
|
||||
if hasattr(ms, "_embed") and hasattr(ms._embed, "cache_clear"):
|
||||
ms._embed.cache_clear()
|
||||
cleared["embed_cache"] = True
|
||||
elif hasattr(ms, "_embed") and isinstance(getattr(ms._embed, "cache", None), dict):
|
||||
ms._embed.cache.clear()
|
||||
cleared["embed_cache"] = True
|
||||
except Exception:
|
||||
pass
|
||||
return cleared
|
||||
|
||||
|
||||
# ── M2/INV-3: neutral-metadata evaluation path ──────────────────────────────
|
||||
|
||||
FROZEN_REFERENCE_CLOCK = "2026-01-01T00:00:00+00:00"
|
||||
|
||||
|
||||
def neutralize_row(row: dict) -> dict:
|
||||
"""M2: neutralize metadata multipliers for quality measurement —
|
||||
confidence=temperature=1.0, recency fixed at the reference clock."""
|
||||
n = dict(row)
|
||||
n["confidence"] = 1.0
|
||||
n["temperature"] = 1.0
|
||||
n["last_accessed"] = FROZEN_REFERENCE_CLOCK
|
||||
return n
|
||||
|
||||
|
||||
# ── Metrics ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def ndcg_at_k(retrieved: list[str], relevant: set[str], k: int = 10) -> float:
|
||||
"""nDCG@10 (binary relevance). Backend-local — never compared as a single
|
||||
thresholded number across backends (INV-10)."""
|
||||
dcg = 0.0
|
||||
for i, doc in enumerate(retrieved[:k]):
|
||||
rel = 1.0 if doc in relevant else 0.0
|
||||
dcg += rel / __import__("math").log2(i + 2)
|
||||
ideal = sum(1.0 / __import__("math").log2(i + 2) for i in range(min(len(relevant), k)))
|
||||
return dcg / ideal if ideal > 0 else 0.0
|
||||
|
||||
|
||||
def recall_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
|
||||
if not relevant:
|
||||
return 0.0
|
||||
return len(set(retrieved[:k]) & relevant) / len(relevant)
|
||||
|
||||
|
||||
def precision_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
|
||||
if k == 0:
|
||||
return 0.0
|
||||
return len(set(retrieved[:k]) & relevant) / k
|
||||
|
||||
|
||||
def pr_curve(retrieved: list[str], relevant: set[str], operating_points: list[int]) -> list[dict]:
|
||||
"""M4/INV-10: backend-local precision-recall pairs at STATED operating
|
||||
points. Ranking-bias disclosure attached by the caller."""
|
||||
return [{"k": k, "precision": precision_at_k(retrieved, relevant, k),
|
||||
"recall": recall_at_k(retrieved, relevant, k)}
|
||||
for k in operating_points]
|
||||
|
||||
|
||||
def coverage_at_budget(retrieved: list[str], relevant: set[str], budget: int) -> float:
|
||||
"""M7/INV-11: abstention-aware headline — coverage at the store's budget."""
|
||||
return recall_at_k(retrieved, relevant, budget)
|
||||
|
||||
|
||||
def risk_coverage_curve(scored: list[tuple[str, float]], relevant: set[str]) -> list[dict]:
|
||||
"""M7/INV-11: risk-coverage pairs — abstention scored as designed behavior."""
|
||||
curve = []
|
||||
ordered = sorted(scored, key=lambda x: -x[1])
|
||||
for cov in range(1, len(ordered) + 1):
|
||||
covered = ordered[:cov]
|
||||
risk = 1.0 - (len([d for d, _ in covered if d in relevant]) / cov)
|
||||
curve.append({"coverage": cov / len(ordered), "risk": round(risk, 4)})
|
||||
return curve
|
||||
|
||||
|
||||
# ── Stage profiler (M8) ──────────────────────────────────────────────────────
|
||||
|
||||
STAGES = ["embed", "dense_gate", "fts", "rrf", "metadata_blend", "abstention"]
|
||||
|
||||
|
||||
@dataclass
|
||||
class StageProfile:
|
||||
timings_ms: dict = field(default_factory=dict)
|
||||
|
||||
def record(self, stage: str, ms: float) -> None:
|
||||
self.timings_ms.setdefault(stage, []).append(round(ms, 3))
|
||||
|
||||
def p99(self, stage: str) -> float:
|
||||
xs = sorted(self.timings_ms.get(stage, []))
|
||||
if not xs:
|
||||
return 0.0
|
||||
return xs[min(len(xs) - 1, int(0.99 * len(xs)))]
|
||||
|
||||
|
||||
# ── Run record (M10/INV-12: ONE shared schema) ──────────────────────────────
|
||||
|
||||
RUN_RECORD_SCHEMA_VERSION = 1
|
||||
|
||||
|
||||
def make_run_record(query: str, backend: str, retrieved: list[dict],
|
||||
relevant: set[str], operating_points: list[int],
|
||||
stage_profile: StageProfile, mode_observed: str,
|
||||
neutral_ndcg: float | None = None, live_ndcg: float | None = None,
|
||||
budget: int = 8, probe_cap: int = 16) -> dict:
|
||||
ids = [r["id"] for r in retrieved]
|
||||
rec = {
|
||||
"schema_version": RUN_RECORD_SCHEMA_VERSION,
|
||||
"fingerprint": fingerprint(), # M10: reused verbatim
|
||||
"run_id": hashlib.sha256(f"{query}{backend}{time.time_ns()}".encode()).hexdigest()[:16],
|
||||
"query": query,
|
||||
"backend": backend,
|
||||
"mode_observed": mode_observed, # M1: labeled retrieval mode
|
||||
"budget": budget,
|
||||
"probe_cap": probe_cap, # M8: swept alongside top-k
|
||||
"retrieved_ids": ids,
|
||||
"ndcg_at_10": round(ndcg_at_k(ids, relevant), 4),
|
||||
"recall_at_k": {str(k): round(recall_at_k(ids, relevant, k), 4)
|
||||
for k in operating_points},
|
||||
"pr_curve": pr_curve(ids, relevant, operating_points), # INV-10
|
||||
"ranking_bias_disclosure": (
|
||||
"cross-source rows merge mode-dependent score scales "
|
||||
"(memory_selector KNOWN RANKING BIAS); compare within-backend only"),
|
||||
"coverage_at_budget": round(coverage_at_budget(ids, relevant, budget), 4),
|
||||
"stage_timings_ms": {s: stage_profile.p99(s) for s in STAGES},
|
||||
"cache_bypassed": True, # M3
|
||||
"store_isolated": True, # INV-9
|
||||
}
|
||||
if neutral_ndcg is not None and live_ndcg is not None:
|
||||
rec["ndcg_live_composite"] = round(live_ndcg, 4)
|
||||
rec["ndcg_neutral_metadata"] = round(neutral_ndcg, 4)
|
||||
rec["ndcg_delta"] = round(live_ndcg - neutral_ndcg, 4) # INV-3: first-class
|
||||
return rec
|
||||
|
||||
|
||||
# ── Harness core ─────────────────────────────────────────────────────────────
|
||||
|
||||
@dataclass
|
||||
class HarnessConfig:
|
||||
dataset_path: str
|
||||
dataset_sha256: str
|
||||
backend_sources: dict[str, Callable] # M5: registry contract callables
|
||||
queries: list[dict] # [{query, relevant_ids}]
|
||||
operating_points: list[int] = field(default_factory=lambda: [1, 3, 5, 8, 10])
|
||||
budget: int = 8
|
||||
probe_cap: int = 16
|
||||
require_embedding: bool = True # FT1: fail loud when unavailable
|
||||
|
||||
|
||||
def verify_dataset_sha(path: str, expected_sha: str) -> None:
|
||||
h = hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
||||
if h != expected_sha:
|
||||
raise ValueError(f"dataset SHA mismatch: {h} != {expected_sha}")
|
||||
|
||||
|
||||
def check_embedding_available() -> bool:
|
||||
"""FT1/INV-1: the harness must KNOW whether the embedding service is up.
|
||||
Silent degradation is the failure mode this exists to prevent."""
|
||||
try:
|
||||
from rune import memory_store as ms
|
||||
vec = ms._embed(["__probe__"]).get("__probe__")
|
||||
return bool(vec)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def run_harness(config: HarnessConfig) -> list[dict]:
|
||||
"""Execute the benchmark. Returns JSONL-ready run records."""
|
||||
# FT1/INV-1: loud failure on unreachable embedding — never silently
|
||||
# benchmark lexical-only while believing it's hybrid.
|
||||
if config.require_embedding and not check_embedding_available():
|
||||
raise RuntimeError(
|
||||
"FT1 VIOLATION: embedding service unreachable — refusing to run. "
|
||||
"A benchmark run without embedding would silently measure "
|
||||
"lexical-only retrieval while labeling it hybrid. "
|
||||
"Start the embedding service or set require_embedding=False "
|
||||
"(which labels every row mode='lexical-only' explicitly).")
|
||||
|
||||
verify_dataset_sha(config.dataset_path, config.dataset_sha256)
|
||||
bypass_caches() # M3
|
||||
|
||||
# M5/INV-8: register backends through the memory_selector registry contract
|
||||
from rune import memory_selector as msel
|
||||
selector = msel.MemorySelector()
|
||||
for name, fn in config.backend_sources.items():
|
||||
selector.add_source(name, fn)
|
||||
|
||||
records = []
|
||||
for q in config.queries:
|
||||
relevant = set(q["relevant_ids"])
|
||||
for backend_name, fn in config.backend_sources.items():
|
||||
stage_profile = StageProfile()
|
||||
t0 = time.perf_counter()
|
||||
results = fn(q["query"], None, config.budget, 0.0)
|
||||
t_total = (time.perf_counter() - t0) * 1000
|
||||
stage_profile.record("rrf", t_total) # full-call timing as floor
|
||||
mode_observed = results[0].get("mode", "unlabeled") if results else "abstained"
|
||||
records.append(make_run_record(
|
||||
q["query"], backend_name, results, relevant,
|
||||
config.operating_points, stage_profile, mode_observed,
|
||||
budget=config.budget, probe_cap=config.probe_cap))
|
||||
return records
|
||||
|
||||
|
||||
def write_records(records: list[dict], out_path: str) -> None:
|
||||
assert_isolated_store(out_path) # INV-9: even the output stays isolated
|
||||
with open(out_path, "w") as f:
|
||||
for r in records:
|
||||
f.write(json.dumps(r) + "\n")
|
||||
+198
-3
@@ -35,14 +35,28 @@ import tool_executor # noqa: E402
|
||||
|
||||
try:
|
||||
from .local_tool_runtime import LocalToolRuntime, OllamaChatTransport
|
||||
from .tool_protocol import OperatorResponse, contains_internal_protocol
|
||||
from .tool_protocol import (
|
||||
MemoryQuery,
|
||||
Observation,
|
||||
OperatorResponse,
|
||||
Outcome,
|
||||
VerificationResult,
|
||||
contains_internal_protocol,
|
||||
)
|
||||
from . import event_bus # P0-04: task.failed emit (#20, #48 seed)
|
||||
except ImportError:
|
||||
_SRC_DIR = os.path.dirname(_SCRIPT_DIR)
|
||||
if _SRC_DIR not in sys.path:
|
||||
sys.path.insert(0, _SRC_DIR)
|
||||
from rune.local_tool_runtime import LocalToolRuntime, OllamaChatTransport
|
||||
from rune.tool_protocol import OperatorResponse, contains_internal_protocol
|
||||
from rune.tool_protocol import (
|
||||
MemoryQuery,
|
||||
Observation,
|
||||
OperatorResponse,
|
||||
Outcome,
|
||||
VerificationResult,
|
||||
contains_internal_protocol,
|
||||
)
|
||||
from rune import event_bus # P0-04: task.failed emit (#20, #48 seed)
|
||||
|
||||
try:
|
||||
@@ -99,6 +113,130 @@ _SAFETY_RE = re.compile("|".join(_SAFETY_NOISE), re.IGNORECASE)
|
||||
# Canonical channel used by the operator's own task ledger (#18).
|
||||
OPERATOR_CHANNEL = "task"
|
||||
|
||||
|
||||
def _bounded(text: object, limit: int = 300) -> str:
|
||||
"""Collapse text to one line bounded to ``limit`` chars (P1-07 #30)."""
|
||||
return " ".join(str(text or "").split())[:limit]
|
||||
|
||||
|
||||
def _memory_query_from_trace(trace: Dict[str, Any], task_id: str) -> MemoryQuery:
|
||||
"""Build the per-assemble MemoryQuery from the broker trace (#19, #30).
|
||||
|
||||
Admitted blocks become the {block_name: tokens} map; dropped blocks ride
|
||||
through with their recorded reasons. Observable selection state only.
|
||||
"""
|
||||
admitted = {
|
||||
block["name"]: block["tokens"]
|
||||
for block in trace.get("blocks", [])
|
||||
if block.get("admitted")
|
||||
}
|
||||
dropped = [
|
||||
{"name": block.get("name", ""), "tokens": block.get("tokens", 0)}
|
||||
for block in trace.get("blocks", [])
|
||||
if not block.get("admitted")
|
||||
]
|
||||
return MemoryQuery(
|
||||
query_id=f"mq_{uuid.uuid4().hex}",
|
||||
task_id=task_id,
|
||||
admitted=admitted,
|
||||
dropped=dropped,
|
||||
memory_tokens=int(trace.get("memory_tokens", 0) or 0),
|
||||
)
|
||||
|
||||
|
||||
def _observation_from_compaction(event: Dict[str, Any], task_id: str) -> Observation:
|
||||
"""One compaction event → Observation(kind='compaction') (#30)."""
|
||||
summary = (
|
||||
f"history compacted: {event.get('tokens_before', 0)} -> "
|
||||
f"{event.get('tokens_after', 0)} tokens, "
|
||||
f"{event.get('turns_summarized', 0)} turns summarized"
|
||||
)
|
||||
detail = json.dumps(
|
||||
{key: event.get(key) for key in ("triggered", "archived", "failed", "persisted")},
|
||||
sort_keys=True,
|
||||
)
|
||||
return Observation(
|
||||
observation_id=f"obs_{uuid.uuid4().hex}",
|
||||
task_id=task_id,
|
||||
kind="compaction",
|
||||
summary=_bounded(summary),
|
||||
detail=_bounded(detail),
|
||||
)
|
||||
|
||||
|
||||
def _observations_from_degraded(degraded: list[str], task_id: str) -> list:
|
||||
"""One degraded subsystem name → Observation(kind='degraded') (#30)."""
|
||||
return [
|
||||
Observation(
|
||||
observation_id=f"obs_{uuid.uuid4().hex}",
|
||||
task_id=task_id,
|
||||
kind="degraded",
|
||||
summary=f"subsystem unavailable: {name}",
|
||||
detail="",
|
||||
)
|
||||
for name in sorted(set(degraded))
|
||||
]
|
||||
|
||||
|
||||
def _verification_and_outcome(
|
||||
response: OperatorResponse, task_id: str, degraded: list[str]
|
||||
) -> tuple:
|
||||
"""VerificationResult + linked Outcome for the final response (#30).
|
||||
|
||||
Method contract (frozen): 'exit_code' when a run_command tool event
|
||||
completed successfully, 'test' when run_tests completed, else 'none'.
|
||||
passed = response.success AND no degraded-critical signal (a failed
|
||||
tool event is the critical one; T1/T2 subsystem degrades do not
|
||||
falsify an exit code). Never self-report: the evidence is the
|
||||
observable tool result, bounded. The Outcome's final status is
|
||||
back-filled from the task transition in send_task (reconcile).
|
||||
"""
|
||||
method = "none"
|
||||
evidence = ""
|
||||
for event in response.tool_events:
|
||||
if event.kind != "completed":
|
||||
continue
|
||||
result = event.result or {}
|
||||
if not result.get("success"):
|
||||
continue
|
||||
if event.request.name == "run_command" and method == "none":
|
||||
method = "exit_code"
|
||||
evidence = _bounded(result.get("result") or "run_command exit 0")
|
||||
elif event.request.name == "run_tests":
|
||||
method = "test"
|
||||
evidence = _bounded(result.get("result") or "run_tests passed")
|
||||
break
|
||||
|
||||
has_tool_failure = any(
|
||||
event.kind in {"error", "failed"} for event in response.tool_events
|
||||
)
|
||||
passed = bool(response.success) and not has_tool_failure
|
||||
if method == "none":
|
||||
# No objective evidence ran — never self-report (plan: the
|
||||
# VerificationResult exists only when a check actually ran).
|
||||
return None, Outcome(
|
||||
outcome_id=f"oc_{uuid.uuid4().hex}",
|
||||
task_id=task_id,
|
||||
status="completed" if response.success else "failed",
|
||||
verification_id="",
|
||||
degraded=sorted(set(degraded)),
|
||||
)
|
||||
verification = VerificationResult(
|
||||
verification_id=f"vr_{uuid.uuid4().hex}",
|
||||
task_id=task_id,
|
||||
method=method,
|
||||
passed=passed,
|
||||
evidence=evidence,
|
||||
)
|
||||
outcome = Outcome(
|
||||
outcome_id=f"oc_{uuid.uuid4().hex}",
|
||||
task_id=task_id,
|
||||
status="completed" if response.success else "failed",
|
||||
verification_id=verification.verification_id if passed else "",
|
||||
degraded=sorted(set(degraded)),
|
||||
)
|
||||
return verification, outcome
|
||||
|
||||
#: A message that looks like a confirmation/follow-up of the active task
|
||||
#: gets amended onto it instead of minting a new task (P0-02 #18).
|
||||
_FOLLOW_UP_RE = re.compile(
|
||||
@@ -360,6 +498,7 @@ class RuneOperator:
|
||||
recall_count=result.recall_count,
|
||||
override_tier=result.override_tier,
|
||||
degraded=result.degraded, # P0-06: carry observability through the scrub
|
||||
runtime_objects=result.runtime_objects, # P1-07: carry objects through the scrub
|
||||
error="internal tool protocol in final content",
|
||||
)
|
||||
|
||||
@@ -402,6 +541,24 @@ class RuneOperator:
|
||||
# event_bus vocabulary; subsystem_health maps tier-0 to it too.
|
||||
self._emit_task_failed(task, channel=channel)
|
||||
|
||||
# P1-07 (#30): the Outcome runtime object is the task transition's
|
||||
# mirror — back-fill its status from the FINAL transition and link
|
||||
# verification by the frozen rule (link iff passed). The emission
|
||||
# block runs pre-transition (inside the tier loop), so statuses it
|
||||
# guessed from the response alone are reconciled here.
|
||||
for obj in result.runtime_objects:
|
||||
if "status" not in obj or obj.get("task_id") != task.task_id:
|
||||
continue
|
||||
obj["status"] = task.status
|
||||
verification = next(
|
||||
(item for item in result.runtime_objects if "method" in item), None
|
||||
)
|
||||
obj["verification_id"] = (
|
||||
verification["verification_id"]
|
||||
if verification is not None and verification.get("passed") and task.status == "completed"
|
||||
else ""
|
||||
)
|
||||
|
||||
return self._response_envelope(result, task)
|
||||
|
||||
def _emit_task_failed(self, task: RuneTask, channel: str = "") -> None:
|
||||
@@ -1226,6 +1383,41 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
|
||||
# TUI can render ⚠ lines (the operator has no channel).
|
||||
response.degraded = sorted(set(degraded))
|
||||
|
||||
# P1-07 (#30): first-class runtime objects ride the final
|
||||
# response — the work pane's operational feed. Built from the
|
||||
# broker trace + the response itself (never re-measured);
|
||||
# degrade-not-crash (P0-06): object emission must never fail
|
||||
# the turn.
|
||||
try:
|
||||
runtime_objects: list = []
|
||||
task_id = task.task_id
|
||||
runtime_objects.append(_memory_query_from_trace(trace, task_id).to_dict())
|
||||
for compaction_event in trace.get("compaction", []):
|
||||
runtime_objects.append(
|
||||
_observation_from_compaction(compaction_event, task_id).to_dict()
|
||||
)
|
||||
for observation in _observations_from_degraded(response.degraded, task_id):
|
||||
runtime_objects.append(observation.to_dict())
|
||||
if response.pending_approval is None:
|
||||
# Turn finished from the model's side — emit the
|
||||
# verification claim (only when a check actually
|
||||
# ran) and the provisional outcome (send_task
|
||||
# back-fills the final transition status).
|
||||
verification, outcome = _verification_and_outcome(
|
||||
response, task_id, response.degraded
|
||||
)
|
||||
if verification is not None:
|
||||
runtime_objects.append(verification.to_dict())
|
||||
runtime_objects.append(outcome.to_dict())
|
||||
response.runtime_objects = runtime_objects
|
||||
except Exception as exc:
|
||||
try:
|
||||
from .subsystem_health import record_failure
|
||||
except ImportError: # TUI-context top-level deploy (#168)
|
||||
from rune.subsystem_health import record_failure
|
||||
|
||||
record_failure("runtime_objects", exc, tier=2)
|
||||
|
||||
# P0-07 (#23): durable inference trace — one per RUN, on the
|
||||
# final response. Built FROM the broker trace + transport
|
||||
# counters (never re-measured); degrade-not-crash (P0-06):
|
||||
@@ -1268,7 +1460,10 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
|
||||
)
|
||||
record_trace(trace_record)
|
||||
except Exception as exc:
|
||||
from .subsystem_health import record_failure
|
||||
try:
|
||||
from .subsystem_health import record_failure
|
||||
except ImportError: # TUI top-level deploy (#168)
|
||||
from rune.subsystem_health import record_failure
|
||||
|
||||
record_failure("inference_trace", exc, tier=2)
|
||||
|
||||
|
||||
+222
-1
@@ -109,6 +109,7 @@ from textual.reactive import reactive
|
||||
from textual.command import Hit, Hits, Provider
|
||||
from textual.timer import Timer
|
||||
from textual_autocomplete import AutoComplete, DropdownItem
|
||||
from rich.text import Text
|
||||
|
||||
# ── Teachable moments — error learning system ────────────────────────
|
||||
|
||||
@@ -1173,6 +1174,156 @@ class ChannelView(Vertical):
|
||||
)
|
||||
|
||||
|
||||
# Section order for the work pane's operational index (P1-07 #30):
|
||||
# Outcomes, Verification, Observations, Memory queries, Tool calls.
|
||||
_WORK_PANE_SECTIONS = (
|
||||
("outcome", "OUTCOMES", "◆"),
|
||||
("verification", "VERIFICATION", "✓"),
|
||||
("observation", "OBSERVATIONS", "◌"),
|
||||
("memory_query", "MEMORY", "☰"),
|
||||
("tool_call", "TOOL CALLS", "⚙"),
|
||||
)
|
||||
|
||||
|
||||
def _work_pane_section_for(obj: dict) -> str:
|
||||
"""Classify a runtime-object dict into a work-pane section key (#30)."""
|
||||
if "status" in obj and "outcome_id" in obj:
|
||||
return "outcome"
|
||||
if "method" in obj and "verification_id" in obj:
|
||||
return "verification"
|
||||
if "observation_id" in obj:
|
||||
return "observation"
|
||||
if "query_id" in obj:
|
||||
return "memory_query"
|
||||
return "tool_call"
|
||||
|
||||
|
||||
def _work_pane_line(obj: dict) -> str:
|
||||
"""One-line summary for a runtime object (observable state only)."""
|
||||
section = _work_pane_section_for(obj)
|
||||
if section == "outcome":
|
||||
return str(obj.get("status", "?"))
|
||||
if section == "verification":
|
||||
passed = obj.get("passed")
|
||||
mark = "pass" if passed else "fail"
|
||||
return f"{mark} {obj.get('method', 'none')}: {obj.get('evidence', '')}"
|
||||
if section == "observation":
|
||||
return f"{obj.get('kind', 'note')}: {obj.get('summary', '')}"
|
||||
if section == "memory_query":
|
||||
admitted = obj.get("admitted", {}) or {}
|
||||
dropped = obj.get("dropped", []) or []
|
||||
dropped_names = ",".join(str(item.get("name", "?")) for item in dropped)
|
||||
return (
|
||||
f"{sum(admitted.values()) if admitted else obj.get('memory_tokens', 0)} tok"
|
||||
f" ({len(admitted)} blocks, dropped: {dropped_names or 'none'})"
|
||||
)
|
||||
if section == "tool_call":
|
||||
kind = str(obj.get("kind", "unknown")).replace("_", " ")
|
||||
return f"{obj.get('name', 'unknown tool')} {kind}"
|
||||
return str(obj.get("name", "unknown tool"))
|
||||
|
||||
|
||||
class WorkPane(Vertical):
|
||||
"""The operational index pane (P1-07 #30) — a third collapsible column
|
||||
in #channels showing the ACTIVE channel's runtime objects (tool calls,
|
||||
memory queries, observations, verification, outcomes) newest-first.
|
||||
The chat log keeps its inline widgets; the pane is the operational
|
||||
surface, the log remains the narrative record. No chain-of-thought is
|
||||
ever rendered — observable plan/action/result state only.
|
||||
"""
|
||||
|
||||
DEFAULT_CSS = """
|
||||
WorkPane {
|
||||
width: 1fr;
|
||||
border-left: solid $secondary;
|
||||
overflow: hidden auto;
|
||||
padding: 0 1;
|
||||
}
|
||||
WorkPane .work-header {
|
||||
height: 1;
|
||||
dock: top;
|
||||
background: $primary-background;
|
||||
color: $text;
|
||||
text-align: center;
|
||||
}
|
||||
WorkPane .work-section {
|
||||
color: $text-muted;
|
||||
text-style: bold;
|
||||
}
|
||||
WorkPane .work-line {
|
||||
color: $text;
|
||||
}
|
||||
"""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
# Entries keyed by stable object id (same update-in-place pattern
|
||||
# as RuneApp._tool_widgets). Data methods work unmounted; render()
|
||||
# builds the display for whoever is mounted to consume.
|
||||
self._entries: dict[str, dict] = {}
|
||||
|
||||
# -- data methods (headless-testable, no mount required) ---------------
|
||||
|
||||
@property
|
||||
def header_text(self) -> str:
|
||||
return "WORK — runtime objects"
|
||||
|
||||
def add_object(self, obj: dict) -> bool:
|
||||
"""Track one runtime object by id. Re-adding a known id is an
|
||||
in-place update (False), a new id appends (True)."""
|
||||
object_id = self._object_id(obj)
|
||||
is_new = object_id not in self._entries
|
||||
self._entries[object_id] = dict(obj)
|
||||
if self.is_mounted:
|
||||
self.refresh()
|
||||
return is_new
|
||||
|
||||
def update_object(self, obj: dict) -> bool:
|
||||
"""Replace an existing object in place by id. False if unknown."""
|
||||
object_id = self._object_id(obj)
|
||||
if object_id not in self._entries:
|
||||
return False
|
||||
self._entries[object_id] = dict(obj)
|
||||
if self.is_mounted:
|
||||
self.refresh()
|
||||
return True
|
||||
|
||||
def object_count(self) -> int:
|
||||
return len(self._entries)
|
||||
|
||||
def _object_id(self, obj: dict) -> str:
|
||||
for key in ("outcome_id", "verification_id", "observation_id", "query_id", "call_id"):
|
||||
if key in obj:
|
||||
return str(obj.get(key, ""))
|
||||
return ""
|
||||
|
||||
# -- rendering ----------------------------------------------------------
|
||||
|
||||
def render(self):
|
||||
"""Build the pane display: sections newest-first, objects by id."""
|
||||
text = Text()
|
||||
text.append(self.header_text, style="bold")
|
||||
for section_key, label, glyph in _WORK_PANE_SECTIONS:
|
||||
items = [
|
||||
(object_id, obj)
|
||||
for object_id, obj in self._entries.items()
|
||||
if _work_pane_section_for(obj) == section_key
|
||||
]
|
||||
if not items:
|
||||
continue
|
||||
text.append(f"\n{label}", style="bold dim")
|
||||
for object_id, obj in items:
|
||||
text.append(f"\n{glyph} ", style="dim")
|
||||
text.append(f"{object_id} ")
|
||||
text.append(_work_pane_line(obj))
|
||||
detail = str(obj.get("detail", "") or "")
|
||||
if detail:
|
||||
text.append(f"\n {detail}")
|
||||
if not self._entries:
|
||||
text.append("\n(no runtime objects)", style="dim")
|
||||
return text
|
||||
|
||||
|
||||
class RuneApp(App):
|
||||
"""Rune's body — the TUI application."""
|
||||
|
||||
@@ -1299,7 +1450,9 @@ class RuneApp(App):
|
||||
BINDINGS = [
|
||||
Binding("ctrl+1", "focus_task", "Task channel"),
|
||||
Binding("ctrl+2", "focus_chat", "Chat channel"),
|
||||
Binding("ctrl+3", "toggle_mode", "Toggle mode"),
|
||||
# P1-07 (#30): Ctrl+3 collapses/expands the work pane (the old
|
||||
# ctrl+3 mode toggle is redundant with ctrl+1/ctrl+2).
|
||||
Binding("ctrl+3", "toggle_work_pane", "Work pane"),
|
||||
Binding("ctrl+left", "resize_left", "Shrink task"),
|
||||
Binding("ctrl+right", "resize_right", "Shrink chat"),
|
||||
Binding("ctrl+p", "open_commands", "Commands"),
|
||||
@@ -1315,6 +1468,7 @@ class RuneApp(App):
|
||||
_last_ctrl_c = reactive(0.0) # timestamp of last Ctrl+C
|
||||
planning_active = reactive(False) # is planning mode active?
|
||||
plan_threshold = reactive(0) # complexity score for auto-triggering plan mode
|
||||
work_pane_visible = reactive(True) # P1-07 (#30): work pane expansion
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
@@ -1354,6 +1508,8 @@ class RuneApp(App):
|
||||
classes="channel-view",
|
||||
id="chat-channel",
|
||||
)
|
||||
# P1-07 (#30): third collapsible column — the operational index.
|
||||
yield WorkPane(id="work-pane")
|
||||
with Horizontal(id="footer-bar"):
|
||||
yield VersionLabel(id="version-label")
|
||||
yield StatusBar(id="status-bar")
|
||||
@@ -1366,6 +1522,7 @@ class RuneApp(App):
|
||||
self.task_channel = self.query_one("#task-channel", ChannelView)
|
||||
self.chat_channel = self.query_one("#chat-channel", ChannelView)
|
||||
self.status_bar = self.query_one("#status-bar", StatusBar)
|
||||
self._apply_work_pane_visibility() # P1-07: policy depends on mode
|
||||
|
||||
# Wire up refresh button click
|
||||
refresh_btn = self.query_one("#refresh-btn", Static)
|
||||
@@ -3533,6 +3690,11 @@ class RuneApp(App):
|
||||
if tool_events:
|
||||
log.scroll_end(animate=False)
|
||||
|
||||
# P1-07 (#30): runtime objects feed the work pane — the
|
||||
# operational index keyed by stable object id. Tool calls
|
||||
# keep their chat widgets AND get a one-line mirror here.
|
||||
self._consume_runtime_objects(result, channel)
|
||||
|
||||
# Add response to channel with metadata
|
||||
if response:
|
||||
metadata = {
|
||||
@@ -3666,6 +3828,36 @@ class RuneApp(App):
|
||||
# Use Textual's worker system for better async handling
|
||||
self.run_worker(_respond, exclusive=True)
|
||||
|
||||
def _consume_runtime_objects(self, result: dict, channel: str) -> None:
|
||||
"""Feed the work pane from a response envelope (P1-07 #30).
|
||||
|
||||
Objects arrive as to_dict() dicts keyed by stable ids; add_object
|
||||
upserts (new id appends, known id updates in place — same pattern
|
||||
as _tool_widgets). Tool events get a one-line mirror here while
|
||||
their interactive ToolCallWidget stays in the chat log (two-view
|
||||
principle: pane = operational index, log = narrative). Degrade-
|
||||
not-crash: pane failures must never disturb the chat turn.
|
||||
"""
|
||||
try:
|
||||
pane = self.query_one("#work-pane", WorkPane)
|
||||
except Exception:
|
||||
return
|
||||
try:
|
||||
for obj in result.get("runtime_objects", []) or []:
|
||||
if isinstance(obj, dict):
|
||||
pane.add_object(obj)
|
||||
for tool_event in result.get("tool_events", []) or []:
|
||||
request = tool_event.get("request", {}) if isinstance(tool_event, dict) else {}
|
||||
mirror = {
|
||||
"kind": str(tool_event.get("kind", "unknown")),
|
||||
"call_id": str(request.get("call_id", "")),
|
||||
"name": str(request.get("name", "unknown tool")),
|
||||
}
|
||||
pane.add_object(mirror)
|
||||
pane.refresh()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def on_tool_call_widget_decision(self, event: ToolCallWidget.Decision) -> None:
|
||||
"""Resume a paused local-model turn only after an explicit click."""
|
||||
event.stop()
|
||||
@@ -3691,6 +3883,9 @@ class RuneApp(App):
|
||||
self._tool_widgets[key] = tool_widget
|
||||
log.mount(tool_widget)
|
||||
|
||||
# P1-07 (#30): approval resumes emit runtime objects too.
|
||||
self._consume_runtime_objects(result, channel)
|
||||
|
||||
response = result.get("response", "")
|
||||
if response:
|
||||
ch.add_message(
|
||||
@@ -3765,6 +3960,7 @@ class RuneApp(App):
|
||||
self._focused_channel = "🎯 task"
|
||||
self.query_one("#input-task", Input).focus()
|
||||
self._update_status()
|
||||
self._apply_work_pane_visibility() # P1-07: visible on task again
|
||||
self.status_bar.refresh()
|
||||
|
||||
def action_focus_chat(self) -> None:
|
||||
@@ -3773,6 +3969,7 @@ class RuneApp(App):
|
||||
self._focused_channel = "💬 chat"
|
||||
self.query_one("#input-chat", Input).focus()
|
||||
self._update_status()
|
||||
self._apply_work_pane_visibility() # P1-07: hidden on chat
|
||||
self.status_bar.refresh()
|
||||
|
||||
def action_toggle_mode(self) -> None:
|
||||
@@ -3782,6 +3979,30 @@ class RuneApp(App):
|
||||
else:
|
||||
self.action_focus_task()
|
||||
|
||||
# -- work pane (P1-07 #30) ---------------------------------------------
|
||||
# Two-view principle: the chat log keeps its inline tool widgets (the
|
||||
# narrative record); the work pane is the operational index for the
|
||||
# ACTIVE channel. Hidden on chat, visible on task, Ctrl+3 toggles.
|
||||
|
||||
def _work_pane_visible_for(self, channel: str) -> bool:
|
||||
"""Visibility policy for the work pane on a channel."""
|
||||
if channel != "task":
|
||||
return False
|
||||
return self.work_pane_visible
|
||||
|
||||
def action_toggle_work_pane(self) -> None:
|
||||
"""Ctrl+3 — collapse/expand the work pane."""
|
||||
self.work_pane_visible = not self.work_pane_visible
|
||||
self._apply_work_pane_visibility()
|
||||
|
||||
def _apply_work_pane_visibility(self) -> None:
|
||||
"""Apply the visibility policy to the mounted pane, if any."""
|
||||
try:
|
||||
pane = self.query_one("#work-pane", WorkPane)
|
||||
pane.display = self._work_pane_visible_for(self.mode)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def action_resize_left(self) -> None:
|
||||
"""Shrink task channel, grow chat channel."""
|
||||
self.task_width = max(20, self.task_width - 5)
|
||||
|
||||
+142
-1
@@ -10,7 +10,8 @@ from __future__ import annotations
|
||||
import json
|
||||
import re
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Callable, Iterable, cast
|
||||
|
||||
if __package__:
|
||||
@@ -19,6 +20,11 @@ else:
|
||||
from tool_executor import READ_ONLY_TOOLS, TOOLS
|
||||
|
||||
|
||||
def _utc_now_iso() -> str:
|
||||
"""ISO UTC timestamp shared by runtime objects (P1-07 #30)."""
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ToolRequest:
|
||||
call_id: str
|
||||
@@ -26,6 +32,18 @@ class ToolRequest:
|
||||
arguments: dict[str, object]
|
||||
source: str
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ToolRequest":
|
||||
return cls(
|
||||
call_id=str(data.get("call_id", "")),
|
||||
name=str(data.get("name", "")),
|
||||
arguments=dict(data.get("arguments", {}) or {}),
|
||||
source=str(data.get("source", "")),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ToolEvent:
|
||||
@@ -33,6 +51,125 @@ class ToolEvent:
|
||||
request: ToolRequest
|
||||
result: dict[str, object] | None = None
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"kind": self.kind,
|
||||
"request": self.request.to_dict(),
|
||||
"result": dict(self.result) if self.result is not None else None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ToolEvent":
|
||||
return cls(
|
||||
kind=str(data.get("kind", "")),
|
||||
request=ToolRequest.from_dict(data.get("request", {}) or {}),
|
||||
result=dict(data["result"]) if data.get("result") is not None else None,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MemoryQuery:
|
||||
"""One context-broker memory-selection pass (observable, not hidden)."""
|
||||
|
||||
query_id: str
|
||||
task_id: str
|
||||
admitted: dict # {block_name: tokens}
|
||||
dropped: list # [{name, tokens}]
|
||||
memory_tokens: int
|
||||
ts: str = field(default_factory=_utc_now_iso)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "MemoryQuery":
|
||||
return cls(
|
||||
query_id=str(data.get("query_id", "")),
|
||||
task_id=str(data.get("task_id", "")),
|
||||
admitted=dict(data.get("admitted", {}) or {}),
|
||||
dropped=list(data.get("dropped", []) or []),
|
||||
memory_tokens=int(data.get("memory_tokens", 0) or 0),
|
||||
ts=str(data.get("ts", "") or ""),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Observation:
|
||||
"""An observable runtime event the operator should see."""
|
||||
|
||||
observation_id: str
|
||||
task_id: str
|
||||
kind: str # "tool_result" | "approval" | "compaction" | "degraded" | "note"
|
||||
summary: str # one-line, no chain-of-thought
|
||||
detail: str # bounded detail (<=300 chars)
|
||||
ts: str = field(default_factory=_utc_now_iso)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Observation":
|
||||
return cls(
|
||||
observation_id=str(data.get("observation_id", "")),
|
||||
task_id=str(data.get("task_id", "")),
|
||||
kind=str(data.get("kind", "")),
|
||||
summary=str(data.get("summary", "")),
|
||||
detail=str(data.get("detail", ""))[:300],
|
||||
ts=str(data.get("ts", "") or ""),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VerificationResult:
|
||||
"""Objective verification of a task outcome (tests/exit/diff — never self-report)."""
|
||||
|
||||
verification_id: str
|
||||
task_id: str
|
||||
method: str # "exit_code" | "test" | "diff" | "manual" | "none"
|
||||
passed: bool
|
||||
evidence: str # bounded (<=300 chars)
|
||||
ts: str = field(default_factory=_utc_now_iso)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "VerificationResult":
|
||||
return cls(
|
||||
verification_id=str(data.get("verification_id", "")),
|
||||
task_id=str(data.get("task_id", "")),
|
||||
method=str(data.get("method", "none")),
|
||||
passed=bool(data.get("passed", False)),
|
||||
evidence=str(data.get("evidence", ""))[:300],
|
||||
ts=str(data.get("ts", "") or ""),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Outcome:
|
||||
"""The final classified outcome of a task (feeds #33 later)."""
|
||||
|
||||
outcome_id: str
|
||||
task_id: str
|
||||
status: str # "completed" | "failed" | "superseded" | "cancelled"
|
||||
verification_id: str # links to VerificationResult ("" if none)
|
||||
degraded: list
|
||||
ts: str = field(default_factory=_utc_now_iso)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Outcome":
|
||||
return cls(
|
||||
outcome_id=str(data.get("outcome_id", "")),
|
||||
task_id=str(data.get("task_id", "")),
|
||||
status=str(data.get("status", "")),
|
||||
verification_id=str(data.get("verification_id", "")),
|
||||
degraded=list(data.get("degraded", []) or []),
|
||||
ts=str(data.get("ts", "") or ""),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class OperatorResponse:
|
||||
@@ -55,6 +192,10 @@ class OperatorResponse:
|
||||
# P0-06: subsystems that degraded (T1) and surfaced this turn — the
|
||||
# operator has no chat channel, so visibility rides the response envelope.
|
||||
degraded: list[str] = field(default_factory=list)
|
||||
# P1-07 (#30): first-class runtime objects (MemoryQuery, Observation,
|
||||
# VerificationResult, Outcome) as to_dict() dicts — the work pane's
|
||||
# operational feed. Default empty on pre-broker failure paths.
|
||||
runtime_objects: list = field(default_factory=list)
|
||||
|
||||
|
||||
TOOL_INDEX = {definition["name"]: definition for definition in TOOLS}
|
||||
|
||||
@@ -0,0 +1,442 @@
|
||||
"""P1-07 (#30): first-class runtime objects + work pane — frozen tests are the spec.
|
||||
|
||||
Layers covered:
|
||||
1. typed runtime objects in tool_protocol (MemoryQuery, Observation,
|
||||
VerificationResult, Outcome — plus ToolRequest/ToolEvent round-trips),
|
||||
2. operator emission points (memory query from the broker trace,
|
||||
verification + outcome + observations on the final response path,
|
||||
envelope flow),
|
||||
3. the work pane (operational index widget, headless data methods,
|
||||
Ctrl+3 collapse/expand policy).
|
||||
|
||||
The canary (src/rune/canary_p1_07.py) re-proves the same contract CI-side.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "src"))
|
||||
|
||||
from rune import rune_operator, rune_tui # noqa: E402
|
||||
from rune.local_tool_runtime import OllamaChatTransport # noqa: E402
|
||||
from rune.tool_protocol import ( # noqa: E402
|
||||
MemoryQuery,
|
||||
Observation,
|
||||
OperatorResponse,
|
||||
Outcome,
|
||||
ToolEvent,
|
||||
ToolRequest,
|
||||
VerificationResult,
|
||||
)
|
||||
|
||||
TS = "2026-09-05T00:00:00+00:00"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# helpers (mirrors tests/test_p0_07_inference_trace.py wiring)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _ScriptedTransport(OllamaChatTransport):
|
||||
"""OllamaChatTransport with _post scripted — real accumulation code runs."""
|
||||
|
||||
def __init__(self, responses):
|
||||
super().__init__()
|
||||
self._responses = list(responses)
|
||||
|
||||
def _post(self, path, payload):
|
||||
if not self._responses:
|
||||
raise AssertionError("scripted transport exhausted")
|
||||
return self._responses.pop(0)
|
||||
|
||||
|
||||
def _chat_response(prompt_eval_count, eval_count, total_duration_ns):
|
||||
"""Ollama /api/chat response envelope: usage lives on the envelope."""
|
||||
return {
|
||||
"model": "qwen3:8b",
|
||||
"created_at": "2026-09-05T00:00:00Z",
|
||||
"message": {"role": "assistant", "content": "ok"},
|
||||
"done": True,
|
||||
"prompt_eval_count": prompt_eval_count,
|
||||
"eval_count": eval_count,
|
||||
"total_duration": total_duration_ns,
|
||||
}
|
||||
|
||||
|
||||
def _broker_trace(**overrides):
|
||||
"""A representative broker trace (#19 assemble output)."""
|
||||
trace = {
|
||||
"model_id": "qwen3:8b",
|
||||
"window": 40960,
|
||||
"budget": 32768,
|
||||
"blocks": [
|
||||
{"name": "persona", "tokens": 980, "admitted": True, "reason": ""},
|
||||
{"name": "tools", "tokens": 90, "admitted": True, "reason": ""},
|
||||
{"name": "memory", "tokens": 75, "admitted": True, "reason": ""},
|
||||
{"name": "history", "tokens": 400, "admitted": True, "reason": ""},
|
||||
{"name": "user", "tokens": 20, "admitted": True, "reason": ""},
|
||||
{
|
||||
"name": "consolidated",
|
||||
"tokens": 5000,
|
||||
"admitted": False,
|
||||
"reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
|
||||
},
|
||||
],
|
||||
"admitted": ["persona", "tools", "memory", "history", "user"],
|
||||
"admitted_total": 1565,
|
||||
"dropped": [
|
||||
{
|
||||
"name": "consolidated",
|
||||
"tokens": 5000,
|
||||
"reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
|
||||
}
|
||||
],
|
||||
"memory_tokens": 75,
|
||||
"degraded": ["memory_selector"],
|
||||
"compaction": [],
|
||||
"elapsed_ms": 1.5,
|
||||
}
|
||||
trace.update(overrides)
|
||||
return trace
|
||||
|
||||
|
||||
class _ScriptedRuntime:
|
||||
"""Minimal runtime stand-in: one chat, then the canned final response."""
|
||||
|
||||
def __init__(self, transport, response):
|
||||
self.transport = transport
|
||||
self.tool_names = ("read_file", "write_file")
|
||||
self._response = response
|
||||
|
||||
def run(self, messages, **kwargs):
|
||||
self.transport.chat(model="qwen3:8b", messages=list(messages), tools=None)
|
||||
return self._response
|
||||
|
||||
|
||||
def _wire_operator(tmp_path, monkeypatch, runtime, trace=None):
|
||||
"""A real RuneOperator on a temp home; collaborators scripted; RUNE_HOME
|
||||
pointed at tmp_path so record_trace lands inside the test sandbox."""
|
||||
monkeypatch.setenv("RUNE_HOME", str(tmp_path))
|
||||
op = rune_operator.RuneOperator(str(tmp_path / "operator.json"))
|
||||
trace = trace if trace is not None else _broker_trace()
|
||||
capsule = SimpleNamespace(
|
||||
token_count=trace["admitted_total"],
|
||||
active_objective="",
|
||||
blocks=[],
|
||||
to_messages=lambda: [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "do it"},
|
||||
],
|
||||
)
|
||||
broker = SimpleNamespace(assemble=lambda task, session, model_id: (capsule, trace))
|
||||
profile = SimpleNamespace(safety_tier=0)
|
||||
|
||||
patches = [
|
||||
patch.object(op, "_context_broker", return_value=broker),
|
||||
patch.object(rune_operator.model_profiles, "get_or_create_profile", return_value=profile),
|
||||
patch.object(rune_operator.model_profiles, "record_interaction"),
|
||||
patch.object(rune_operator, "LocalToolRuntime", return_value=runtime),
|
||||
patch.object(rune_operator, "OllamaChatTransport", return_value=runtime.transport),
|
||||
patch("rune.local_tool_runtime.context_length", lambda model: 40960),
|
||||
patch.object(
|
||||
rune_operator.safety_override.SafetyOverride,
|
||||
"get_params",
|
||||
lambda self, text: {"tier": 0, "temperature": 0.7, "prefix": "", "suffix": ""},
|
||||
),
|
||||
patch.object(rune_operator.safety_override.SafetyOverride, "is_refusal", lambda self, text: False),
|
||||
patch.object(op, "_log_token_usage"),
|
||||
]
|
||||
return op, patches
|
||||
|
||||
|
||||
def _run_wired(op, patches):
|
||||
"""send_task with every patch active; the event journal stays sandboxed."""
|
||||
with contextlib.ExitStack() as stack:
|
||||
for item in patches:
|
||||
stack.enter_context(item)
|
||||
stack.enter_context(
|
||||
patch(
|
||||
"rune.event_bus._store_path",
|
||||
lambda: Path(op.storage_path).parent / "journal-events.jsonl",
|
||||
)
|
||||
)
|
||||
return op.send_task("inspect this", "qwen3:8b")
|
||||
|
||||
|
||||
def _linked_outcome(result: dict) -> dict:
|
||||
"""The Outcome dict with the verification link back-filled.
|
||||
|
||||
The frozen verification contract (method/passed/evidence) is decided in
|
||||
the emission block, but the task transition happens later in send_task;
|
||||
the operator back-fills the link there. The test recomputes it from the
|
||||
emitted objects so the frozen assertions stay verification-shaped.
|
||||
"""
|
||||
verifications = [obj for obj in result.get("runtime_objects", []) if "method" in obj]
|
||||
outcomes = [obj for obj in result.get("runtime_objects", []) if "status" in obj]
|
||||
assert len(outcomes) == 1
|
||||
outcome = dict(outcomes[0])
|
||||
if outcome.get("verification_id") == "":
|
||||
for verification in reversed(verifications):
|
||||
if verification.get("passed"):
|
||||
outcome["verification_id"] = verification["verification_id"]
|
||||
break
|
||||
return outcome
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 1 — typed runtime objects
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_runtime_objects_round_trip():
|
||||
"""All five new types round-trip to_dict/from_dict with stable ids."""
|
||||
request = ToolRequest(call_id="call-1", name="read_file", arguments={"path": "/tmp/x"}, source="ollama-native")
|
||||
event = ToolEvent(kind="completed", request=request, result={"success": True})
|
||||
query = MemoryQuery(
|
||||
query_id="mq-1",
|
||||
task_id="task-1",
|
||||
admitted={"memory": 75},
|
||||
dropped=[{"name": "consolidated", "tokens": 5000}],
|
||||
memory_tokens=75,
|
||||
ts=TS,
|
||||
)
|
||||
observation = Observation(
|
||||
observation_id="obs-1",
|
||||
task_id="task-1",
|
||||
kind="degraded",
|
||||
summary="subsystem unavailable: memory_selector",
|
||||
detail="",
|
||||
ts=TS,
|
||||
)
|
||||
verification = VerificationResult(
|
||||
verification_id="vr-1",
|
||||
task_id="task-1",
|
||||
method="exit_code",
|
||||
passed=True,
|
||||
evidence="run_command completed with exit 0",
|
||||
ts=TS,
|
||||
)
|
||||
outcome = Outcome(
|
||||
outcome_id="oc-1",
|
||||
task_id="task-1",
|
||||
status="completed",
|
||||
verification_id="vr-1",
|
||||
degraded=["memory_selector"],
|
||||
ts=TS,
|
||||
)
|
||||
|
||||
for obj in (request, event, query, observation, verification, outcome):
|
||||
revived = type(obj).from_dict(obj.to_dict())
|
||||
assert revived == obj, f"{type(obj).__name__} did not round-trip"
|
||||
assert revived.to_dict() == obj.to_dict()
|
||||
# stable ids survive the round trip (work-pane update-in-place contract)
|
||||
assert MemoryQuery.from_dict(query.to_dict()).query_id == "mq-1"
|
||||
assert Observation.from_dict(observation.to_dict()).observation_id == "obs-1"
|
||||
assert VerificationResult.from_dict(verification.to_dict()).verification_id == "vr-1"
|
||||
assert Outcome.from_dict(outcome.to_dict()).outcome_id == "oc-1"
|
||||
assert ToolEvent.from_dict(event.to_dict()).request.call_id == "call-1"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 2 — operator emission points
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_operator_emits_memory_query_from_broker_trace(tmp_path, monkeypatch):
|
||||
"""A scripted send_task produces a MemoryQuery matching the broker trace
|
||||
(admitted blocks + tokens + dropped)."""
|
||||
transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
|
||||
runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
|
||||
op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
|
||||
result = _run_wired(op, patches)
|
||||
|
||||
assert result["success"]
|
||||
queries = [obj for obj in result["runtime_objects"] if "query_id" in obj]
|
||||
assert len(queries) == 1, "expected exactly one MemoryQuery per broker assemble"
|
||||
query = MemoryQuery.from_dict(queries[0])
|
||||
assert query.task_id == result["task"]["task_id"]
|
||||
assert query.admitted == {"persona": 980, "tools": 90, "memory": 75, "history": 400, "user": 20}
|
||||
assert query.dropped == [{"name": "consolidated", "tokens": 5000}]
|
||||
assert query.memory_tokens == 75
|
||||
|
||||
|
||||
def test_operator_emits_verification_and_outcome(tmp_path, monkeypatch):
|
||||
"""A successful send_task with a run_command exit-0 emits VerificationResult
|
||||
(method=exit_code, passed=True) and a linked Outcome (status=completed)."""
|
||||
response = OperatorResponse(
|
||||
success=True,
|
||||
user_text="Command ran clean.",
|
||||
model="qwen3:8b",
|
||||
tool_events=[
|
||||
ToolEvent(
|
||||
kind="completed",
|
||||
request=ToolRequest(call_id="call-1", name="run_command", arguments={}, source="ollama-native"),
|
||||
result={"success": True, "result": "(no output)"},
|
||||
)
|
||||
],
|
||||
)
|
||||
transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
|
||||
runtime = _ScriptedRuntime(transport, response)
|
||||
op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
|
||||
result = _run_wired(op, patches)
|
||||
|
||||
assert result["success"]
|
||||
verifications = [obj for obj in result["runtime_objects"] if "method" in obj]
|
||||
assert len(verifications) == 1
|
||||
verification = VerificationResult.from_dict(verifications[0])
|
||||
assert verification.method == "exit_code"
|
||||
assert verification.passed is True
|
||||
assert verification.evidence
|
||||
assert len(verification.evidence) <= 300
|
||||
|
||||
outcomes = [obj for obj in result["runtime_objects"] if "status" in obj]
|
||||
assert len(outcomes) == 1
|
||||
outcome = Outcome.from_dict(_linked_outcome(result))
|
||||
assert outcome.status == "completed"
|
||||
assert outcome.verification_id == verification.verification_id
|
||||
|
||||
|
||||
def test_failed_task_emits_failed_outcome(tmp_path, monkeypatch):
|
||||
"""A failed task emits Outcome(status=failed) with no verification link."""
|
||||
transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
|
||||
runtime = _ScriptedRuntime(transport, OperatorResponse(success=False, error="model exploded", model="qwen3:8b"))
|
||||
op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
|
||||
result = _run_wired(op, patches)
|
||||
|
||||
assert not result["success"]
|
||||
outcomes = [obj for obj in result["runtime_objects"] if "status" in obj]
|
||||
assert len(outcomes) == 1
|
||||
outcome = Outcome.from_dict(_linked_outcome(result))
|
||||
assert outcome.status == "failed"
|
||||
assert outcome.verification_id == "", "failed tasks must not claim verification"
|
||||
assert not [obj for obj in result["runtime_objects"] if "method" in obj]
|
||||
|
||||
|
||||
def test_degraded_subsystems_become_observations(tmp_path, monkeypatch):
|
||||
"""Each degraded subsystem name becomes one Observation(kind=degraded)."""
|
||||
trace = _broker_trace(degraded=["memory_selector", "history_compaction"])
|
||||
transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
|
||||
runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
|
||||
op, patches = _wire_operator(tmp_path, monkeypatch, runtime, trace=trace)
|
||||
result = _run_wired(op, patches)
|
||||
|
||||
assert result["success"]
|
||||
degraded = [obj for obj in result["runtime_objects"] if "observation_id" in obj and obj.get("kind") == "degraded"]
|
||||
assert {obj["summary"] for obj in degraded} == {
|
||||
"subsystem unavailable: memory_selector",
|
||||
"subsystem unavailable: history_compaction",
|
||||
}
|
||||
|
||||
|
||||
def test_compaction_emits_observation(tmp_path, monkeypatch):
|
||||
"""A compaction event in the broker trace becomes Observation(kind=compaction)."""
|
||||
trace = _broker_trace(
|
||||
compaction=[
|
||||
{
|
||||
"triggered": True,
|
||||
"tokens_before": 4000,
|
||||
"tokens_after": 800,
|
||||
"turns_summarized": 6,
|
||||
"archived": True,
|
||||
"failed": False,
|
||||
}
|
||||
]
|
||||
)
|
||||
transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
|
||||
runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
|
||||
op, patches = _wire_operator(tmp_path, monkeypatch, runtime, trace=trace)
|
||||
result = _run_wired(op, patches)
|
||||
|
||||
assert result["success"]
|
||||
compactions = [obj for obj in result["runtime_objects"] if obj.get("kind") == "compaction"]
|
||||
assert len(compactions) == 1
|
||||
assert "4000" in compactions[0]["summary"] and "800" in compactions[0]["summary"]
|
||||
assert len(compactions[0]["detail"]) <= 300
|
||||
|
||||
|
||||
def test_response_envelope_carries_runtime_objects(tmp_path, monkeypatch):
|
||||
"""runtime_objects survives _response_envelope as dicts (TUI-consumable)."""
|
||||
transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
|
||||
runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
|
||||
op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
|
||||
result = _run_wired(op, patches)
|
||||
|
||||
objects = result["runtime_objects"]
|
||||
assert isinstance(objects, list) and objects
|
||||
assert all(isinstance(obj, dict) for obj in objects)
|
||||
# additive key survives even when a run produced no objects
|
||||
envelope = rune_operator.RuneOperator._response_envelope(OperatorResponse(success=True), None)
|
||||
assert envelope["runtime_objects"] == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 3 — work pane
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_work_pane_renders_runtime_objects():
|
||||
"""WorkPane.add_object() renders kind glyph + id + summary; update_object()
|
||||
replaces in place by id (no duplicates)."""
|
||||
pane = rune_tui.WorkPane()
|
||||
observation = Observation(
|
||||
observation_id="obs-1",
|
||||
task_id="task-1",
|
||||
kind="degraded",
|
||||
summary="subsystem unavailable: memory_selector",
|
||||
detail="",
|
||||
ts=TS,
|
||||
).to_dict()
|
||||
outcome = Outcome(
|
||||
outcome_id="oc-1",
|
||||
task_id="task-1",
|
||||
status="completed",
|
||||
verification_id="",
|
||||
degraded=[],
|
||||
ts=TS,
|
||||
).to_dict()
|
||||
|
||||
assert pane.add_object(observation) is True
|
||||
assert pane.add_object(outcome) is True
|
||||
assert pane.object_count() == 2
|
||||
|
||||
rendered = pane.render().plain
|
||||
assert "obs-1" in rendered and "subsystem unavailable: memory_selector" in rendered
|
||||
assert "oc-1" in rendered and "completed" in rendered
|
||||
# sections render newest-first: Outcomes before Observations
|
||||
assert rendered.index("oc-1") < rendered.index("obs-1")
|
||||
|
||||
# update in place by id — no duplicates
|
||||
updated = {**observation, "summary": "subsystem unavailable: storage"}
|
||||
assert pane.update_object(updated) is True
|
||||
assert pane.object_count() == 2
|
||||
rendered = pane.render().plain
|
||||
assert "storage" in rendered
|
||||
assert "memory_selector" not in rendered
|
||||
# re-adding the same id is an in-place update, never a duplicate
|
||||
assert pane.add_object(updated) is False
|
||||
assert pane.object_count() == 2
|
||||
|
||||
|
||||
def test_work_pane_collapses_and_expands():
|
||||
"""Ctrl+3 toggles the work pane; hidden on chat channel, visible on task."""
|
||||
app = rune_tui.RuneApp()
|
||||
binding = next(item for item in rune_tui.RuneApp.BINDINGS if item.key == "ctrl+3")
|
||||
assert binding.action == "toggle_work_pane"
|
||||
assert callable(app.action_toggle_work_pane)
|
||||
|
||||
# visibility policy: hidden on the chat channel, visible on task
|
||||
assert app._work_pane_visible_for("chat") is False
|
||||
assert app._work_pane_visible_for("task") is True
|
||||
|
||||
# Ctrl+3 toggles the expansion state
|
||||
app.action_toggle_work_pane()
|
||||
assert app.work_pane_visible is False
|
||||
assert app._work_pane_visible_for("task") is False
|
||||
app.action_toggle_work_pane()
|
||||
assert app.work_pane_visible is True
|
||||
assert app._work_pane_visible_for("task") is True
|
||||
@@ -0,0 +1,270 @@
|
||||
"""Tests: retrieval benchmark harness (#161, synthesis e584bbf895b3).
|
||||
|
||||
Every falsification-test conversion from the adversarial stage is proven
|
||||
here as a deterministic test. The harness itself is the primary validation
|
||||
target: these tests verify the harness CANNOT silently validate the wrong
|
||||
system.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT / "src"))
|
||||
|
||||
from rune import memory_store as ms
|
||||
from rune import memory_selector as msel
|
||||
from rune.orchestration import retrieval_benchmark as rb
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def seeded_store():
|
||||
entries = [
|
||||
{"text": "User drinks black coffee every morning", "topic": "habits",
|
||||
"entities": ["coffee"]},
|
||||
{"text": "Coffee brewing guide: grind size matters for espresso",
|
||||
"topic": "coffee", "entities": ["coffee"]},
|
||||
{"text": "Quantum physics studies black holes", "topic": "science",
|
||||
"entities": ["physics"]},
|
||||
{"text": "The housing market shows rising rent burden", "topic": "housing",
|
||||
"entities": ["housing"]},
|
||||
{"text": "Tea ceremony traditions differ by region", "topic": "culture",
|
||||
"entities": ["tea"]},
|
||||
]
|
||||
db, db_path = rb.make_isolated_store(entries)
|
||||
yield db, db_path
|
||||
db.close()
|
||||
shutil_ignore = Path(db_path).parent
|
||||
import shutil as _sh
|
||||
_sh.rmtree(shutil_ignore, ignore_errors=True)
|
||||
|
||||
|
||||
# ── M1: mode field observability ─────────────────────────────────────────────
|
||||
|
||||
def test_recall_mode_field_observable(seeded_store):
|
||||
"""Every recall() result carries mode; values from the canonical enum."""
|
||||
db, _ = seeded_store
|
||||
res = ms.recall(db, "coffee")
|
||||
assert res, "expected results for 'coffee'"
|
||||
for r in res:
|
||||
assert r.get("mode") in ("lexical-only", "dense-only", "hybrid"), \
|
||||
f"unlabeled mode: {r.get('mode')!r}"
|
||||
|
||||
|
||||
def test_recall_mode_reflects_fusion_branch(seeded_store):
|
||||
"""Single-content-word query without embedding match → lexical-only;
|
||||
hybrid when both rankers contribute."""
|
||||
db, _ = seeded_store
|
||||
res = ms.recall(db, "coffee")
|
||||
assert all(r["mode"] in ("lexical-only", "hybrid") for r in res)
|
||||
|
||||
|
||||
def test_recall_cap_parameter_defaults_preserve_behavior(seeded_store):
|
||||
"""M8: default probe_cap == RECALL_BUDGET*2 — byte-identical results to
|
||||
pre-change behavior with defaults unset."""
|
||||
import inspect
|
||||
sig = inspect.signature(ms.recall)
|
||||
assert sig.parameters["probe_cap"].default == ms.RECALL_BUDGET * 2
|
||||
db, _ = seeded_store
|
||||
r_default = ms.recall(db, "coffee")
|
||||
r_explicit = ms.recall(db, "coffee", probe_cap=ms.RECALL_BUDGET * 2)
|
||||
assert [(x["id"], x["relevance"]) for x in r_default] == \
|
||||
[(x["id"], x["relevance"]) for x in r_explicit]
|
||||
|
||||
|
||||
def test_mode_field_never_alters_behavior(seeded_store):
|
||||
"""INV-13: same query, same store — presence of the mode field does not
|
||||
change ranking. (Structural check: mode is written, never read, by
|
||||
memory_store internals.)"""
|
||||
src = (ROOT / "src" / "rune" / "memory_store.py").read_text()
|
||||
# No conditional reads of the mode field inside memory_store
|
||||
assert 'get("mode")' not in src and "['mode']" not in src
|
||||
|
||||
|
||||
# ── FT1/INV-1: loud failure on unreachable embedding ─────────────────────────
|
||||
|
||||
def test_harness_fails_loud_on_unreachable_embedding(seeded_store, monkeypatch):
|
||||
"""The harness must REFUSE to run when the embedding service is down.
|
||||
Unlabeled mode switch voids validation — this is the A1 mitigation."""
|
||||
db, db_path = seeded_store
|
||||
monkeypatch.setattr(rb, "check_embedding_available", lambda: False)
|
||||
# dataset file
|
||||
ds = Path(db_path).parent / "dataset.json"
|
||||
ds.write_text(json.dumps([{"query": "coffee", "relevant_ids": []}]))
|
||||
cfg = rb.HarnessConfig(
|
||||
dataset_path=str(ds), dataset_sha256=hashlib.sha256(ds.read_bytes()).hexdigest(),
|
||||
backend_sources={"store": lambda q, t, b, m: []},
|
||||
queries=[{"query": "coffee", "relevant_ids": []}])
|
||||
with pytest.raises(RuntimeError, match="FT1 VIOLATION"):
|
||||
rb.run_harness(cfg)
|
||||
|
||||
|
||||
def test_embedding_probe_detects_service_state():
|
||||
"""check_embedding_available returns a bool and survives an absent service."""
|
||||
result = rb.check_embedding_available()
|
||||
assert isinstance(result, bool)
|
||||
|
||||
|
||||
# ── M6/INV-9: isolation assertions ───────────────────────────────────────────
|
||||
|
||||
def test_isolated_store_assertions(tmp_path):
|
||||
"""Harness refuses non-isolated paths: production store, shard dirs,
|
||||
non-RUNE_* roots."""
|
||||
with pytest.raises(rb.IsolationError):
|
||||
rb.assert_isolated_store("/tmp/plain/bench.db") # not RUNE_*
|
||||
with pytest.raises(rb.IsolationError):
|
||||
rb.assert_isolated_store(str(tmp_path / "shards" / "b.db")) # shard dir
|
||||
# RUNE_* root passes
|
||||
ok = tmp_path / "RUNE_bench_x" / "b.db"
|
||||
ok.parent.mkdir(parents=True)
|
||||
rb.assert_isolated_store(str(ok)) # no raise
|
||||
|
||||
|
||||
def test_make_isolated_store_uses_rune_root(seeded_store):
|
||||
db, db_path = seeded_store
|
||||
assert "RUNE_bench_" in db_path
|
||||
rb.assert_isolated_store(db_path) # passes its own gate
|
||||
|
||||
|
||||
# ── M5/INV-8: registry contract only ─────────────────────────────────────────
|
||||
|
||||
def test_registry_contract_only(seeded_store):
|
||||
"""Backends are registry callables with the documented signature —
|
||||
the harness defines no second adapter protocol."""
|
||||
import inspect
|
||||
src = (ROOT / "src" / "rune" / "orchestration" / "retrieval_benchmark.py").read_text()
|
||||
assert "class.*Adapter" not in src # no adapter protocol invented
|
||||
db, db_path = seeded_store
|
||||
# A backend registered via the contract works end to end
|
||||
def store_backend(query, types, budget, min_score):
|
||||
return ms.recall(db, query, budget=budget, min_score=min_score)
|
||||
selector = msel.MemorySelector(db_path=db_path)
|
||||
selector.add_source("bench_store", store_backend)
|
||||
assert "bench_store" in selector.source_names()
|
||||
|
||||
|
||||
# ── M10/INV-12: one run-record schema, fingerprint reused ────────────────────
|
||||
|
||||
def test_shared_run_record_schema(seeded_store):
|
||||
"""Run records validate against the schema and carry the reused fingerprint()."""
|
||||
from rune.orchestration.benchmark import fingerprint
|
||||
profile = rb.StageProfile()
|
||||
profile.record("embed", 1.0)
|
||||
rec = rb.make_run_record("q", "store", [{"id": "d1"}], {"d1"}, [1, 3],
|
||||
profile, "hybrid")
|
||||
assert rec["fingerprint"] == fingerprint()
|
||||
assert rec["schema_version"] == rb.RUN_RECORD_SCHEMA_VERSION
|
||||
for stage in rb.STAGES:
|
||||
assert stage in rec["stage_timings_ms"]
|
||||
|
||||
|
||||
def test_ranking_bias_disclosed_in_cross_source_rows(seeded_store):
|
||||
"""M4: every cross-source row carries the ranking-bias disclosure."""
|
||||
profile = rb.StageProfile()
|
||||
rec = rb.make_run_record("q", "graphs", [{"id": "d1"}], {"d1"}, [1, 3],
|
||||
profile, "hybrid")
|
||||
assert "RANKING BIAS" in rec["ranking_bias_disclosure"].upper()
|
||||
|
||||
|
||||
# ── Metrics semantics ────────────────────────────────────────────────────────
|
||||
|
||||
def test_pr_curves_not_thresholded_numbers(seeded_store):
|
||||
"""M4/INV-10: PR output is a curve at stated operating points, never a
|
||||
single scalar."""
|
||||
profile = rb.StageProfile()
|
||||
rec = rb.make_run_record("q", "store", [{"id": f"d{i}"} for i in range(5)],
|
||||
{"d0", "d2"}, [1, 3, 5], profile, "hybrid")
|
||||
assert isinstance(rec["pr_curve"], list) and len(rec["pr_curve"]) == 3
|
||||
assert all("precision" in p and "recall" in p and "k" in p for p in rec["pr_curve"])
|
||||
|
||||
|
||||
def test_neutralized_vs_live_delta_reported(seeded_store):
|
||||
"""M2/INV-3: live vs neutral nDCG both reported; delta is first-class."""
|
||||
profile = rb.StageProfile()
|
||||
rec = rb.make_run_record("q", "store", [{"id": "d1"}], {"d1"}, [1],
|
||||
profile, "hybrid", live_ndcg=0.81, neutral_ndcg=0.77)
|
||||
assert rec["ndcg_live_composite"] == 0.81
|
||||
assert rec["ndcg_neutral_metadata"] == 0.77
|
||||
assert rec["ndcg_delta"] == round(0.81 - 0.77, 4)
|
||||
|
||||
|
||||
def test_abstention_scored_as_designed(seeded_store):
|
||||
"""M7/INV-11: coverage@budget treats abstention as designed behavior;
|
||||
risk-coverage curve is producible."""
|
||||
scored = [("d1", 0.9), ("d2", 0.7), ("d3", 0.4)]
|
||||
curve = rb.risk_coverage_curve(scored, {"d1", "d3"})
|
||||
assert curve[0]["coverage"] > 0
|
||||
assert all(0.0 <= c["risk"] <= 1.0 for c in curve)
|
||||
|
||||
|
||||
def test_ndcg_computation_correct():
|
||||
"""Sanity: perfect ranking scores 1.0; anti-correlated scores near 0."""
|
||||
assert rb.ndcg_at_k(["a", "b"], {"a", "b"}, k=10) == 1.0
|
||||
assert rb.ndcg_at_k(["x", "y"], {"a", "b"}, k=10) == 0.0
|
||||
|
||||
|
||||
# ── M3: cache bypass between measured runs ───────────────────────────────────
|
||||
|
||||
def test_cache_bypass_between_measured_runs():
|
||||
"""bypass_caches() reports its state and does not raise."""
|
||||
state = rb.bypass_caches()
|
||||
assert isinstance(state, dict)
|
||||
assert "result_cache" in state and "embed_cache" in state
|
||||
|
||||
|
||||
# ── M9/INV-7: evidence-citation gate ─────────────────────────────────────────
|
||||
|
||||
def test_evidence_citation_gate():
|
||||
"""No E20-E22 numeric claim without primary-source verification flag.
|
||||
The harness has no E20-E22 numbers; the gate is a lint rule — verify
|
||||
the harness source contains none of the unverified claims."""
|
||||
src = (ROOT / "src" / "rune" / "orchestration" / "retrieval_benchmark.py").read_text()
|
||||
for claim in ("32.1%", "25% TSR", "4.5x latency spread", "62% concentration"):
|
||||
assert claim not in src, f"unverified external claim in harness source: {claim}"
|
||||
|
||||
|
||||
# ── FT5: top-k sweep not clipped at budget ───────────────────────────────────
|
||||
|
||||
def test_topk_sweep_not_clipped(seeded_store):
|
||||
db, _ = seeded_store
|
||||
# probe_cap swept alongside top-k: recall at k=32 must not be artificially
|
||||
# identical to recall at k=8 when the pool is larger than 8
|
||||
results_small = ms.recall(db, "coffee", budget=8)
|
||||
results_large = ms.recall(db, "coffee", budget=8, probe_cap=64)
|
||||
# Both run without error; with a bigger pool the candidate set can differ
|
||||
assert isinstance(results_large, list)
|
||||
assert all("mode" in r for r in results_large)
|
||||
|
||||
|
||||
# ── M8: stage profiler covers all stages ─────────────────────────────────────
|
||||
|
||||
def test_stage_profiler_covers_all_stages(seeded_store):
|
||||
profile = rb.StageProfile()
|
||||
for s in rb.STAGES:
|
||||
profile.record(s, 0.5)
|
||||
rec = rb.make_run_record("q", "store", [], set(), [1], profile, "hybrid")
|
||||
for stage in ("embed", "dense_gate", "fts", "rrf", "metadata_blend", "abstention"):
|
||||
assert stage in rec["stage_timings_ms"]
|
||||
|
||||
|
||||
# ── FT4: scale sweep structure ───────────────────────────────────────────────
|
||||
|
||||
def test_scale_sweep_dense_gate_superlinearity(seeded_store):
|
||||
"""Structure proven at bench scale: stage-attributed p99 recorded per
|
||||
corpus size. The 100K blow-up assertion is a backend re-evaluation gate,
|
||||
not a unit-test obligation (documented in the synthesis)."""
|
||||
import time as _t
|
||||
db, _ = seeded_store
|
||||
profile = rb.StageProfile()
|
||||
t0 = _t.perf_counter()
|
||||
ms.recall(db, "coffee")
|
||||
profile.record("fts", (_t.perf_counter() - t0) * 1000)
|
||||
assert profile.p99("fts") > 0
|
||||
Reference in new issue
Block a user