diff --git a/tests/control_plane/test_stride_observation.py b/tests/control_plane/test_stride_observation.py index b56ace572..f5f297a74 100644 --- a/tests/control_plane/test_stride_observation.py +++ b/tests/control_plane/test_stride_observation.py @@ -152,3 +152,244 @@ def test_other_agent_runs_are_not_attributed(tmp_path: Path) -> None: ) assert observation["lineage"]["source_run_count"] == 0 + + +# --- Synthetic stride-boundary fixture (RFC #3204 section 13) ------------- +# +# Hand-built synthetic receipts only: no copied provider payloads, source or +# draft bodies, review text, private locators, host paths, credentials, or +# cursor state. Each assertion traces to one validation criterion of the +# hierarchical stride RFC. + + +def _snapshot_tree(root: Path) -> dict[str, bytes]: + return { + str(path.relative_to(root)): path.read_bytes() + for path in sorted(root.rglob("*")) + if path.is_file() + } + + +def test_boundary_observation_stays_shadow_only_across_segment_states( + tmp_path: Path, +) -> None: + # Criterion 8: shadow mode changes no scheduling, quota, notification, + # gate, or execution behavior. Checked across every synthetic boundary + # state, with a byte snapshot proving the projection never mutates the + # runtime tree it observes. + states: dict[str, list[dict]] = { + "no_receipts": [], + "mid_segment_fresh": [ + _run( + classification="bounded_progress_report", + outcome="outcome_progress", + minutes_ago=90, + ), + _run( + classification="bounded_progress_report", + outcome="outcome_progress", + minutes_ago=30, + ), + ], + "after_authority_marker": [ + _run( + classification="bounded_progress_report", + outcome="outcome_progress", + minutes_ago=120, + ), + _run( + classification="bounded_replan_progress", + outcome="outcome_progress", + minutes_ago=60, + ), + ], + "stale_evidence": [ + _run( + classification="monitor_watch", + outcome="surface_only", + minutes_ago=12 * 60, + ), + ], + } + for name, rows in states.items(): + runtime_root = tmp_path / name + if rows: + _write_run_index(runtime_root, rows) + before = _snapshot_tree(runtime_root) + + observation = build_stride_observation( + runtime_root, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) + evaluation = evaluate_stride_observation(observation) + + assert observation["shadow_only"] is True, name + assert observation["effect"]["unknown"] is True, name + assert evaluation["shadow_only"] is True, name + assert evaluation["recommendations"] == [], name + assert _snapshot_tree(runtime_root) == before, name + + +def test_boundary_missing_metrics_stay_unknown_not_inferred_from_prose( + tmp_path: Path, +) -> None: + # Criterion 2: missing host detail remains unknown rather than inferred + # from prose. Authority-marker words and progress claims live only in + # fields the projection never reads; they must not manufacture metrics. + _write_run_index( + tmp_path, + [ + { + "generated_at": ( + datetime.now(UTC) - timedelta(minutes=30) + ).isoformat(), + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "summary": ( + "operator said this replan changed the vision and the " + "gate; feels like real progress" + ), + }, + { + "generated_at": ( + datetime.now(UTC) - timedelta(minutes=10) + ).isoformat(), + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "classification": "", + "delivery_outcome": "", + "notes": "another stride boundary crossed in prose only", + }, + ], + ) + + observation = build_stride_observation( + tmp_path, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) + + assert observation["lineage"]["source_run_count"] == 2 + assert observation["delivery"]["material_slices"] == 0 + assert observation["delivery"]["latest_outcome"] == "none" + assert observation["authority"]["bounded_slices_since_change"] == 2 + assert observation["authority"]["segment_disposition"] == "unknown" + assert observation["effect"]["unknown"] is True + + +def test_boundary_authority_changes_require_explicit_markers( + tmp_path: Path, +) -> None: + # Criterion 6: reports with no authority delta are not counted as heavy + # steering. Only explicit replan/vision/gate classifications restart the + # bounded-slice count; status, monitor, and review classifications never + # do, and the latest explicit marker wins. + _write_run_index( + tmp_path / "with_markers", + [ + _run( + classification="bounded_progress_report", + outcome="outcome_progress", + minutes_ago=240, + ), + _run( + classification="surface_status_note", + outcome="surface_only", + minutes_ago=180, + ), + _run( + classification="bounded_replan_progress", + outcome="outcome_progress", + minutes_ago=120, + ), + _run( + classification="monitor_watch", + outcome="surface_only", + minutes_ago=60, + ), + _run( + classification="vision_refresh_accepted", + outcome="outcome_progress", + minutes_ago=30, + ), + _run( + classification="exact_head_review_delivered", + outcome="outcome_progress", + minutes_ago=5, + ), + ], + ) + + marked = build_stride_observation( + tmp_path / "with_markers", + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) + # Latest explicit marker sits at index 4; only the final review follows. + assert marked["authority"]["bounded_slices_since_change"] == 1 + + _write_run_index( + tmp_path / "without_markers", + [ + _run( + classification="surface_status_note", + outcome="surface_only", + minutes_ago=120, + ), + _run( + classification="monitor_watch", + outcome="surface_only", + minutes_ago=60, + ), + _run( + classification="exact_head_review_delivered", + outcome="outcome_progress", + minutes_ago=10, + ), + ], + ) + + unmarked = build_stride_observation( + tmp_path / "without_markers", + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) + assert unmarked["authority"]["bounded_slices_since_change"] == 3 + + +def test_boundary_projection_replays_byte_identical(tmp_path: Path) -> None: + # Criterion 3: deterministic replay produces the same stride observation + # and mismatch classification. Identical synthetic receipts are built + # twice in two different runtime roots, with timestamps far from the + # evidence-freshness boundary so wall-clock drift cannot flip a signal. + rows = [ + _run( + classification="bounded_replan_progress", + outcome="outcome_progress", + minutes_ago=300, + ), + _run( + classification="bounded_progress_report", + outcome="outcome_progress", + minutes_ago=45, + ), + ] + roots = [tmp_path / "runtime_alpha", tmp_path / "runtime_beta"] + for root in roots: + _write_run_index(root, rows) + + projections = set() + for root in roots: + for _ in range(2): + observation = build_stride_observation( + root, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) + evaluation = evaluate_stride_observation(observation) + projections.add( + json.dumps([observation, evaluation], sort_keys=True) + ) + + assert len(projections) == 1