Nick
2ac2c5e3c8
CI Pipeline / unit-tests (push) Successful in 28s
CI Pipeline / canary (push) Failing after 24s
CI Pipeline / best-practices (push) Successful in 2s
CI Pipeline / documentation (push) Successful in 2s
CI Pipeline / type-check (push) Successful in 20s
CI Pipeline / gitea-offline (push) Successful in 13s
CI Pipeline / lint (pull_request) Successful in 21s
CI Pipeline / unit-tests (pull_request) Successful in 28s
CI Pipeline / canary (pull_request) Failing after 23s
CI Pipeline / best-practices (pull_request) Successful in 3s
CI Pipeline / documentation (pull_request) Successful in 2s
CI Pipeline / type-check (pull_request) Successful in 20s
CI Pipeline / gitea-offline (pull_request) Successful in 13s
CI Pipeline / required-checks (pull_request) Skipped
CI Pipeline / lint (push) Successful in 28s
CI Pipeline / required-checks (push) Skipped
- I-2: resolve_pending_approval now emits Outcome/Verification/degraded observations via shared _attach_runtime_objects (approved tool calls were executed work but rendered nothing in the pane). - I-3: WorkPane growth capped at 120 entries (oldest evicted); objects stamped with channel; visible_entries() exposes channel-scoped view. - I-1: test asserts the envelope's verification_id equals the emitted verification's id directly (no recompute masking send_task back-fill). - Red→green: 3 new frozen tests failed on missing behavior first (approval-resume emitted no objects; cap/channel absent).
569 lines
23 KiB
Python
569 lines
23 KiB
Python
"""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 shutil
|
|
import uuid
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Quality-gate discharges (PR #180 review) — approved-resume + crash-tier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_approval_resume_emits_outcome(tmp_path, monkeypatch):
|
|
"""Quality I-2: an approval-resumed turn must emit an Outcome object.
|
|
|
|
resolve_pending_approval applies the approved operation and finalizes;
|
|
the resume envelope must carry the same first-class Outcome a direct
|
|
turn gets (approved tool calls are real executed work).
|
|
"""
|
|
transport = _ScriptedTransport([])
|
|
tool_request = ToolRequest(call_id="call-9", name="write_file", arguments={}, source="ollama-native")
|
|
tool_event = ToolEvent(kind="completed", request=tool_request, result={"success": True, "result": "written"})
|
|
resumed = OperatorResponse(
|
|
success=True,
|
|
user_text="Approved and written.",
|
|
model="qwen3:8b",
|
|
tool_events=[tool_event],
|
|
override_tier=0,
|
|
recall_count=0,
|
|
)
|
|
runtime = _ScriptedRuntime(transport, resumed)
|
|
# runtime.approve is what resolve_pending_approval calls
|
|
runtime.approve = lambda pending_state: resumed # type: ignore[method-assign]
|
|
op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
|
|
|
|
# Seed a pending approval on the operator state (the scripted runtime
|
|
# returns a terminal response, so no pending approval exists naturally).
|
|
seeded = rune_operator.RuneTask(task_id=f"task_{uuid.uuid4().hex[:8]}", description="seeded approval")
|
|
op.state.tasks.append(seeded)
|
|
op.state.pending_task_id = seeded.task_id
|
|
op.state.pending_request = {"call_id": "call-9", "name": "write_file", "arguments": {}}
|
|
from rune.tool_executor import PreparedTool, _canonical_digest, _path_digest, serialize_prepared_tool
|
|
import time as _t
|
|
import os as _os
|
|
|
|
# write_file policy allows only /home/nick and /mnt/workdrive — use the
|
|
# same sandbox pattern the P0-05 E2E uses (cleared before use).
|
|
sandbox = Path(_os.path.expanduser("~")) / ".rune-resume-probe"
|
|
import shutil as _shutil
|
|
|
|
_shutil.rmtree(sandbox, ignore_errors=True)
|
|
_shutil.rmtree(sandbox, ignore_errors=True)
|
|
sandbox.mkdir(parents=True)
|
|
target = sandbox / "out.txt"
|
|
approval_token = uuid.uuid4().hex # not a literal credential — henxels wants no token= literal
|
|
prepared = PreparedTool(
|
|
token=approval_token,
|
|
name="write_file",
|
|
arguments={"path": str(target), "content": "x"},
|
|
operation_digest=_canonical_digest("write_file", {"path": str(target), "content": "x"}),
|
|
created_at=_t.time(),
|
|
expires_at=_t.time() + 60,
|
|
requested_path=str(target),
|
|
resolved_path=str(target.resolve()),
|
|
path_digest=_path_digest(target),
|
|
)
|
|
op.state.pending_runtime_state = {"prepared_tool": serialize_prepared_tool(prepared)}
|
|
|
|
with contextlib.ExitStack() as stack:
|
|
for item in patches:
|
|
stack.enter_context(item)
|
|
result = op.resolve_pending_approval(True)
|
|
assert result["success"], result.get("error")
|
|
# probe cleanup: remove the write sandbox the approval touched
|
|
_shutil.rmtree(sandbox, ignore_errors=True)
|
|
outcomes = [obj for obj in result.get("runtime_objects", []) if "status" in obj]
|
|
assert len(outcomes) == 1, "approval-resumed turn must emit exactly one Outcome"
|
|
outcome = Outcome.from_dict(outcomes[0])
|
|
assert outcome.status == "completed"
|
|
assert outcome.task_id == seeded.task_id
|
|
|
|
|
|
def test_failed_verification_backfill_is_straight_read(tmp_path, monkeypatch):
|
|
"""Quality I-1: the outcome's verification link in the ENVELOPE must be
|
|
exactly the emitted verification's id — no test-side recompute masking a
|
|
send_task back-fill regression. (The canary already asserts this; the
|
|
pytest suite is now strictly as strong.)"""
|
|
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)
|
|
|
|
verifications = [obj for obj in result["runtime_objects"] if "method" in obj]
|
|
outcomes = [obj for obj in result["runtime_objects"] if "status" in obj]
|
|
assert len(verifications) == 1 and len(outcomes) == 1
|
|
# Direct envelope assertion — no _linked_outcome recomputation.
|
|
assert outcomes[0]["verification_id"] == verifications[0]["verification_id"]
|
|
assert outcomes[0]["verification_id"] != ""
|
|
|
|
|
|
def test_work_pane_entries_capped_and_channel_scoped():
|
|
"""Quality I-3: pane entries are capped (bounded growth) and the channel
|
|
filter keeps chat objects out of a task-scoped pane view."""
|
|
pane = rune_tui.WorkPane()
|
|
overflow = rune_tui._WORK_PANE_MAX_ENTRIES + 25
|
|
for n in range(overflow):
|
|
pane.add_object({"observation_id": f"obs_{n}", "kind": "degraded", "summary": f"s{n}"})
|
|
assert len(pane._entries) <= rune_tui._WORK_PANE_MAX_ENTRIES
|
|
# newest survive, oldest evicted
|
|
assert "obs_0" not in pane._entries
|
|
assert f"obs_{overflow - 1}" in pane._entries
|
|
# channel scoping: entries carry their channel; visible set filters
|
|
pane2 = rune_tui.WorkPane()
|
|
pane2.add_object({"observation_id": "obs_task", "kind": "degraded", "summary": "t", "channel": "task"})
|
|
pane2.add_object({"observation_id": "obs_chat", "kind": "degraded", "summary": "c", "channel": "chat"})
|
|
visible = pane2.visible_entries("chat")
|
|
assert [oid for oid, _ in visible] == ["obs_chat"]
|