diff --git a/scripts/aoa_session_memory.py b/scripts/aoa_session_memory.py index 830d2665..fb588987 100755 --- a/scripts/aoa_session_memory.py +++ b/scripts/aoa_session_memory.py @@ -138097,6 +138097,113 @@ def compact_goal_lifecycle( } +def goal_lifecycle_task_episode_components( + session_dir: Path, + session_index: dict[str, Any], + lifecycle: dict[str, Any], +) -> dict[str, Any]: + """Select one lifecycle's linked components from the bounded reader. + + The existing manifest-first reader may inspect earlier non-selected shards + while looking for the requested IDs; every selected component is + content-address verified. A bounded selection does not establish + validity of unvisited components or the global task-episode digest. + """ + episode_ids = ( + { + str(item) + for item in lifecycle.get("task_episode_ids", []) + if item + } + if isinstance(lifecycle.get("task_episode_ids"), list) + else set() + ) + requested_count = len(episode_ids) + base = { + "status": "current", + "payloads": [], + "diagnostics": [], + "reader_mode": "not_requested" if not episode_ids else "manifest_first", + "component_count": 0, + "hydrated_component_count": 0, + "selected_component_count": 0, + "scan_complete": requested_count == 0, + "requested_episode_count": requested_count, + "verified_episode_count": 0, + "selected_component_integrity": ( + "not_requested" if not episode_ids else "unresolved" + ), + "global_task_episode_semantic_digest_verified": False, + } + if not episode_ids: + return base + + result = session_index_task_episode_component_read( + session_dir, + session_index, + order="recent", + limit=requested_count, + selector=lambda payload: str(payload.get("episode_id") or "") + in episode_ids, + ) + payloads = [ + dict(item) + for item in result.get("payloads", []) + if isinstance(item, dict) + ] + payload_ids = { + str(item.get("episode_id") or "") + for item in payloads + } + selection_is_exact = ( + result.get("status") == "current" + and len(payloads) == requested_count + and len(payload_ids) == requested_count + and payload_ids == episode_ids + ) + if not selection_is_exact: + diagnostics = [ + str(item) + for item in result.get("diagnostics", []) + if item + ] + if not diagnostics: + diagnostics = [ + "goal_lifecycle_task_episode_component_selection_mismatch" + ] + return { + **base, + "status": "invalid", + "diagnostics": diagnostics, + "reader_mode": str(result.get("reader_mode") or "manifest_first"), + "component_count": int_value(result.get("component_count")), + "hydrated_component_count": int_value( + result.get("hydrated_component_count") + ), + } + return { + **base, + "payloads": payloads, + "reader_mode": str(result.get("reader_mode") or "manifest_first"), + "component_count": int_value(result.get("component_count")), + "hydrated_component_count": int_value( + result.get("hydrated_component_count") + ), + "selected_component_count": len(payloads), + "scan_complete": bool(result.get("scan_complete")), + "verified_episode_count": ( + len(payloads) + if result.get("reader_mode") == "manifest_first" + else 0 + ), + "selected_component_integrity": ( + "verified_content_addressed" + if result.get("reader_mode") == "manifest_first" + else "embedded_legacy_unverified" + ), + } + + def goal_lifecycle_route_search( *, aoa_root: Path, @@ -138130,6 +138237,7 @@ def goal_lifecycle_route_search( results: list[dict[str, Any]] = [] selected_count = 0 incompatible_session_indexes: list[dict[str, Any]] = [] + component_errors: list[dict[str, Any]] = [] ordered_records = list(reversed(records)) if order == "recent" else list(records) normalized_event_kind = route_key_slug(event_kind, fallback="") if event_kind else "" for record in ordered_records: @@ -138165,6 +138273,10 @@ def goal_lifecycle_route_search( generated_session_index_stale_reasons_for_session( session_dir, index, + # The route proves only the selected content-addressed linked + # components below; a global episode digest would hydrate the + # whole component set before lifecycle filtering. + verify_task_episode_semantic_digest=False, ) ) if index_stale_reasons: @@ -138189,32 +138301,69 @@ def goal_lifecycle_route_search( session_label = str(display.get("label") or record.get("session_label") or session_dir.name) session_id = str(index.get("session_id") or record.get("session_id") or "") lifecycles = index.get("goal_lifecycles", []) if isinstance(index.get("goal_lifecycles"), list) else [] - task_episodes = session_index_task_episode_components( - session_dir, - index, - ) lifecycle_items = list(reversed(lifecycles)) if order == "recent" else list(lifecycles) for item in lifecycle_items: if not isinstance(item, dict): continue selected_count += 1 - compact = compact_goal_lifecycle( - item, - session_label=session_label, - session_id=session_id, - aoa_root=aoa_root, - task_episodes=task_episodes, - ) - if goal_id and str(compact.get("goal_id") or "") != goal_id: + if goal_id and str(item.get("goal_id") or "") != goal_id: continue - if status and str(compact.get("status") or "") != status: + if status and str(item.get("status") or "") != status: continue if normalized_event_kind and normalized_event_kind not in { route_key_slug(kind, fallback="") - for kind in compact.get("event_kinds", []) + for kind in ( + item.get("event_kinds", []) + if isinstance(item.get("event_kinds"), list) + else [] + ) if kind }: continue + component_result = goal_lifecycle_task_episode_components( + session_dir, + index, + item, + ) + if component_result.get("status") != "current": + component_errors.append( + { + "session": session_id or session_label, + "goal_id": str(item.get("goal_id") or ""), + "diagnostics": list( + component_result.get("diagnostics", []) + ), + } + ) + continue + compact = compact_goal_lifecycle( + item, + session_label=session_label, + session_id=session_id, + aoa_root=aoa_root, + task_episodes=list(component_result.get("payloads", [])), + ) + compact["task_episode_integrity"] = { + "mode": "bounded_content_addressed_linked_components", + "requested_episode_count": int_value( + component_result.get("requested_episode_count") + ), + "verified_episode_count": int_value( + component_result.get("verified_episode_count") + ), + "selected_component_integrity": str( + component_result.get("selected_component_integrity") + or "unresolved" + ), + "component_count": int_value( + component_result.get("component_count") + ), + "hydrated_component_count": int_value( + component_result.get("hydrated_component_count") + ), + "scan_complete": bool(component_result.get("scan_complete")), + "global_task_episode_semantic_digest_verified": False, + } results.append(compact) if len(results) >= max(1, limit): break @@ -138222,6 +138371,11 @@ def goal_lifecycle_route_search( break provider_status = search_provider_status(aoa_root=aoa_root, provider_name="portable_sqlite") provider_summary = compact_search_provider_status_for_route(provider_status, provider_name="portable_sqlite") + diagnostics: list[str] = [] + if incompatible_session_indexes: + diagnostics.append("goal_lifecycle_source_generation_incompatible") + if component_errors: + diagnostics.append("goal_lifecycle_task_episode_component_invalid") return { "schema_version": SCHEMA_VERSION, "artifact_type": "goal_lifecycle_route_results", @@ -138240,13 +138394,10 @@ def goal_lifecycle_route_search( "incompatible_session_indexes": ( incompatible_session_indexes ), + "component_errors": component_errors, "generation_identity": session_index_generation_identity(), "provider": provider_summary, - "diagnostics": ( - ["goal_lifecycle_source_generation_incompatible"] - if incompatible_session_indexes - else [] - ), + "diagnostics": diagnostics, "next_route": "Use work_chain next_expansion for task/answer context, then refs, graph_refs, raw_refs, and segment_refs for authority; lifecycle packets are generated navigation.", } diff --git a/tests/test_session_memory.py b/tests/test_session_memory.py index f96bb847..0aa89451 100644 --- a/tests/test_session_memory.py +++ b/tests/test_session_memory.py @@ -87929,6 +87929,305 @@ def test_task_episode_cli_route_uses_manifest_first_selective_hydration( ) +def _goal_lifecycle_manifest_fixture( + tmp_path: Path, + *, + include_unselected: bool = True, +) -> tuple[Path, Path, dict[str, Any], dict[str, Any]]: + aoa_root = tmp_path / ".aoa" + session_dir = aoa_root / "sessions" / "goal-route-session" + episodes = [ + { + "episode_id": "task-skipped", + "stable_id": "session:task-skipped:event-1", + "identity": {"canonical_id": "episode:skipped"}, + "status": "closed", + "semantic_text": "skipped episode", + "event_range": {"from_line": 1, "to_line": 1}, + }, + { + "episode_id": "task-selected", + "stable_id": "session:task-selected:event-2", + "identity": {"canonical_id": "episode:selected"}, + "status": "closed", + "semantic_text": "selected episode", + "event_range": {"from_line": 2, "to_line": 2}, + }, + ] if include_unselected else [ + { + "episode_id": "task-selected", + "stable_id": "session:task-selected:event-2", + "identity": {"canonical_id": "episode:selected"}, + "status": "closed", + "semantic_text": "selected episode", + "event_range": {"from_line": 2, "to_line": 2}, + }, + ] + publish_identity = { + "source": { + "raw_sha256": "a" * 64, + "raw_bytes": 200, + "raw_line_count": 2, + }, + "publish_id": "b" * 64, + } + shards = module.materialize_session_index_task_episode_shards( + session_dir, + episodes, + publish_identity=publish_identity, + literal_policy=module.DerivedSessionSensitiveLiteralPolicy( + values_by_kind={} + ), + workers=1, + ) + module.write_session_index_shard_manifest( + session_dir, + publish_identity=publish_identity, + source_identity=shards["source_identity"], + task_episode_records=shards["records"], + ) + + def lifecycle(goal_id: str, episode_id: str) -> dict[str, Any]: + return { + "goal_id": goal_id, + "goal_instance_id": f"{goal_id}-instance", + "status": "complete", + "objective": goal_id, + "objective_source": "test", + "event_count": 0, + "event_kinds": ["goal_completed"], + "event_ids": [], + "task_episode_ids": [episode_id], + "ambiguity_flags": [], + "usage": {}, + "state_observations": [], + "usage_observations": [], + "first_ref": {}, + "last_ref": {}, + "requested_ref": {}, + "failed_ref": {}, + "created_ref": {}, + "completed_ref": {}, + "blocked_ref": {}, + "graph_refs": [], + "raw_refs": [], + "segment_refs": [], + "events": [], + } + + lifecycles = [lifecycle("goal-skipped", "task-skipped")] + if include_unselected: + lifecycles.append(lifecycle("goal-selected", "task-selected")) + session_index = { + "session_id": "goal-route-session-id", + "display": {"label": "goal-route-session"}, + "goal_lifecycles": lifecycles, + "task_episodes": [], + "projection_publish": publish_identity, + "component_storage": { + "manifest_ref": str( + Path(module.SESSION_INDEX_SHARDS_DIR) + / module.SESSION_INDEX_SHARD_MANIFEST_JSON + ), + "task_episode_shard_count": len(episodes), + "embedded_compatibility_view": False, + "reader_mode": "manifest_first", + }, + } + module.write_json(session_dir / module.SESSION_INDEX_JSON, session_index) + record = { + "session_id": "goal-route-session-id", + "session_label": "goal-route-session", + "path": str(session_dir), + } + return aoa_root, session_dir, record, shards + + +def test_goal_lifecycle_route_filters_before_linked_component_hydration( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + aoa_root, _session_dir, record, _shards = _goal_lifecycle_manifest_fixture( + tmp_path + ) + monkeypatch.setattr( + module, + "chronological_session_records", + lambda _aoa_root, limit=None: [record], + ) + monkeypatch.setattr( + module, + "generated_session_index_stale_reasons_from_file", + lambda _path: [], + ) + monkeypatch.setattr( + module, + "generated_session_index_stale_reasons_for_session", + lambda *_args, **_kwargs: [], + ) + monkeypatch.setattr(module, "search_provider_status", lambda **_kwargs: {}) + monkeypatch.setattr( + module, + "compact_search_provider_status_for_route", + lambda *_args, **_kwargs: {}, + ) + calls: list[list[str]] = [] + original = module.goal_lifecycle_task_episode_components + + def observed( + session_dir: Path, + session_index: dict[str, Any], + lifecycle: dict[str, Any], + ) -> dict[str, Any]: + calls.append([str(item) for item in lifecycle["task_episode_ids"]]) + return original(session_dir, session_index, lifecycle) + + monkeypatch.setattr(module, "goal_lifecycle_task_episode_components", observed) + + payload = module.goal_lifecycle_route_search( + aoa_root=aoa_root, + target="all", + goal_id="goal-selected", + limit=1, + ) + + assert payload["ok"] is True + assert payload["result_count"] == 1 + assert calls == [["task-selected"]] + result = payload["results"][0] + assert result["goal_id"] == "goal-selected" + assert result["work_chain"]["episodes"][0]["episode_id"] == ( + "task-selected" + ) + assert result["task_episode_integrity"] == { + "mode": "bounded_content_addressed_linked_components", + "requested_episode_count": 1, + "verified_episode_count": 1, + "selected_component_integrity": "verified_content_addressed", + "component_count": 2, + "hydrated_component_count": 1, + "scan_complete": False, + "global_task_episode_semantic_digest_verified": False, + } + assert payload["component_errors"] == [] + + +def test_goal_lifecycle_route_does_not_compute_global_episode_digest( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + aoa_root, _session_dir, record, _shards = _goal_lifecycle_manifest_fixture( + tmp_path + ) + monkeypatch.setattr( + module, + "chronological_session_records", + lambda _aoa_root, limit=None: [record], + ) + monkeypatch.setattr( + module, + "generated_session_index_stale_reasons_from_file", + lambda _path: [], + ) + original_stale_reasons = module.generated_session_index_stale_reasons_for_session + observed_verify_flags: list[bool] = [] + + def observed_stale_reasons( + session_dir: Path, + session_index: dict[str, Any], + **kwargs: Any, + ) -> list[str]: + observed_verify_flags.append( + bool(kwargs["verify_task_episode_semantic_digest"]) + ) + return original_stale_reasons(session_dir, session_index, **kwargs) + + monkeypatch.setattr( + module, + "generated_session_index_stale_reasons_for_session", + observed_stale_reasons, + ) + monkeypatch.setattr( + module, + "task_episode_semantic_digest_for_session_index", + lambda *_args, **_kwargs: pytest.fail( + "goal lifecycle route must not compute the global episode digest" + ), + ) + monkeypatch.setattr(module, "search_provider_status", lambda **_kwargs: {}) + monkeypatch.setattr( + module, + "compact_search_provider_status_for_route", + lambda *_args, **_kwargs: {}, + ) + + module.goal_lifecycle_route_search( + aoa_root=aoa_root, + target="all", + goal_id="goal-selected", + limit=1, + ) + + assert observed_verify_flags == [False] + + +@pytest.mark.parametrize("tamper", ["missing", "corrupt"]) +def test_goal_lifecycle_route_rejects_invalid_linked_component( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + tamper: str, +) -> None: + aoa_root, session_dir, record, shards = _goal_lifecycle_manifest_fixture( + tmp_path, + include_unselected=False, + ) + shard_path = session_dir / shards["records"][0]["ref"] + if tamper == "missing": + shard_path.unlink() + else: + shard_path.write_text("{}\n", encoding="utf-8") + monkeypatch.setattr( + module, + "chronological_session_records", + lambda _aoa_root, limit=None: [record], + ) + monkeypatch.setattr( + module, + "generated_session_index_stale_reasons_from_file", + lambda _path: [], + ) + monkeypatch.setattr( + module, + "generated_session_index_stale_reasons_for_session", + lambda *_args, **_kwargs: [], + ) + monkeypatch.setattr(module, "search_provider_status", lambda **_kwargs: {}) + monkeypatch.setattr( + module, + "compact_search_provider_status_for_route", + lambda *_args, **_kwargs: {}, + ) + + payload = module.goal_lifecycle_route_search( + aoa_root=aoa_root, + target="all", + goal_id="goal-skipped", + limit=1, + ) + + assert payload["ok"] is True + assert payload["result_count"] == 0 + assert payload["results"] == [] + assert payload["diagnostics"] == [ + "goal_lifecycle_task_episode_component_invalid" + ] + assert payload["component_errors"][0]["goal_id"] == "goal-skipped" + assert any( + reason.startswith("session_index_task_episode_shard_artifact_mismatch") + for reason in payload["component_errors"][0]["diagnostics"] + ) + + def test_navigation_routes_reject_fast_stale_indexes_before_full_json_read( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,