diff --git a/docs/setup-codex.md b/docs/setup-codex.md index 80e441ed..d128be07 100644 --- a/docs/setup-codex.md +++ b/docs/setup-codex.md @@ -72,3 +72,42 @@ truememory-ingest status - **Existing config**: TrueMemory uses additive merges — your existing Codex config is preserved. - **Hooks not firing**: Check that the hook script paths are absolute and the Python executable is correct. - **Windows Defender ASR**: If commands are blocked, use `python -m` form instead. See [debugging guide](guides/debugging.md#windows-defender-asr-blocks-truememory-mcpexe). + +## Native rollout parsing + +The parser supports Codex JSONL and JSON arrays containing native +`response_item` message envelopes. User and assistant text blocks retain file +order and repeated occurrences. Legacy `user_message` / `agent_message` +events and Paginated `item_completed` events are coverage evidence, not extra +conversation turns. Ordinary plain-text lifecycle records, including +`task_started` / `task_complete` and their `turn_*` aliases, are reconciled +against the canonical responses. A terminal summary uses its identified turn +and never supplies another echo credit. + +This subset follows the [pinned Codex producer](https://github.com/openai/codex/tree/0b755b1945bf4a31560df2ce1469aeb044dccbc9/codex-rs). +It excludes explicit harness, skill, AGENTS, tool, inherited, compaction and +retained-delivery context. Positional provenance must align with the original +content array. Explicit unknown provenance cannot authorize text. Older +records without attribution retain role/text compatibility; JSON metadata +does not authenticate authorship or prove that earlier history is present. + +The detailed parser reports partial coverage for unmatched events, ambiguous +lifecycles, unknown variants, media, unavailable delivery and unsupported +assistant rewriting. Citation and proposed-plan markup projections are not +implemented in this subset. Supported canonical text remains available for +inspection, with categorical diagnostics that omit input text and identifiers. +`complete` describes the supported supplied bytes, not the success of the +model's task or an authenticated record of the whole session. + +The stop admission check counts parsed human messages after removing +TrueMemory-injected blocks. The default threshold of five means five nonempty +human occurrences. A long message, metadata, tool results or repeated event +echoes cannot meet that threshold through byte length. Role-marked plain text +still counts; unmarked prose does not. Supported human messages can count in +a partial transcript; admission is not a completion verdict. + +Parser support alone does not establish automatic capture. Native hook +delivery and path admission remain separate integration requirements. Capture +currently uses the list parser API; detailed coverage does not yet control +durable success receipts. The #768 completion-policy work must preserve +partial coverage before automatic-capture completeness can be claimed. diff --git a/tests/test-codex-rollout-762.py b/tests/test-codex-rollout-762.py new file mode 100644 index 00000000..640d3563 --- /dev/null +++ b/tests/test-codex-rollout-762.py @@ -0,0 +1,693 @@ +"""Synthetic native rollout contracts, without importing the application package.""" +from __future__ import annotations + +import ast +import builtins +import copy +import errno +import io +import json +from pathlib import Path +import sys +import types +import unittest +from bisect import bisect_left +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +SENTINEL = "synthetic-private-content" +SCAFFOLD = "synthetic-scaffold-content" +TURN = "synthetic-turn" + + +def load_parser(source: str | None = None) -> types.ModuleType: + module = types.ModuleType("synthetic_codex_parser") + if source is None: + source = (ROOT / "truememory/ingest/transcript.py").read_text(encoding="utf-8") + + def safe_import(name: str, globals: dict | None = None, locals: dict | None = None, + fromlist: tuple = (), level: int = 0) -> object: + if name.startswith(("truememory", "torch", "numpy", "sentence_transformers")): + raise AssertionError("Application imports are forbidden in parser tests") + return builtins.__import__(name, globals, locals, fromlist, level) + + module.__dict__["__builtins__"] = dict(vars(builtins), __import__=safe_import) + with patch.dict(sys.modules, {module.__name__: module}): + exec(compile(source, "synthetic-parser", "exec"), module.__dict__) + return module + + +def response(role: str = "user", parts: tuple[str, ...] = (SENTINEL,), *, + turn: str | None = TURN, phase: str | None = None, item: str = "synthetic-item") -> dict: + payload = {"type": "message", "role": role, "id": item, + "content": [{"type": "input_text" if role == "user" else "output_text", "text": text} + for text in parts]} + if turn is not None: + payload["internal_chat_message_metadata_passthrough"] = {"turn_id": turn} + if phase is not None: + payload["phase"] = phase + return {"type": "response_item", "timestamp": "2026-01-01T00:00:00Z", "payload": payload} + + +def event(kind: str, **fields: object) -> dict: + return {"type": "event_msg", "payload": {"type": kind, **fields}} + + +def start(turn: str = TURN) -> dict: + return event("task_started", turn_id=turn, collaboration_mode_kind="default") + + +def terminal(text: object = "done", turn: str = TURN) -> dict: + return event("task_complete", turn_id=turn, last_agent_message=text) + + +def completed(role: str, text: str, *, turn: str = TURN, + item: str = "synthetic-item", phase: str | None = None) -> dict: + value = {"type": "UserMessage" if role == "user" else "AgentMessage", "id": item, + "content": [{"type": "text" if role == "user" else "Text", "text": text}]} + if phase is not None: + value["phase"] = phase + return event("item_completed", turn_id=turn, thread_id="synthetic-thread", item=value) + + +def lifecycle(*, paginated: bool = False, parts: tuple[str, ...] = ("done",), + phase: str | None = None, turn: str = TURN) -> list[dict]: + user = response(turn=turn) + assistant = response("assistant", parts, turn=turn, phase=phase) + if paginated: + user_echo = completed("user", SENTINEL, turn=turn) + agent_echo = completed("assistant", "".join(parts), turn=turn, phase=phase) + else: + user_echo = event("user_message", message=SENTINEL, images=[], audio=None) + agent_echo = event("agent_message", message="".join(parts), phase=phase) + return [start(turn), user, user_echo, agent_echo, assistant, terminal("".join(parts), turn)] + + +def annotation(kind: str, **fields: object) -> dict: + return {"provenance": {"type": kind, **fields}} + + +def annotate(record: dict, annotations: object = None, kinds: object = None) -> dict: + metadata = record["payload"].setdefault("internal_chat_message_metadata_passthrough", {}) + metadata.update(content_item_metadata=annotations, content_item_kinds=kinds) + return record + + +def load_stop_count(parser: types.ModuleType, text: str, source: str | None = None) -> tuple[object, list[Path]]: + source = source or (ROOT / "truememory/ingest/hooks/stop.py").read_text(encoding="utf-8") + node = next(node for node in ast.parse(source).body + if isinstance(node, ast.FunctionDef) and node.name == "_has_enough_messages") + calls: list[Path] = [] + + def parse(path: Path) -> object: + if not isinstance(path, Path): + raise AssertionError("Stop must select the explicit path contract") + calls.append(path) + return parser._parse_transcript_text_outcome(text.strip()) + + fake = types.SimpleNamespace(parse_transcript_outcome=parse, strip_truememory_blocks=parser.strip_truememory_blocks) + + def safe_import(name: str, globals: dict | None = None, locals: dict | None = None, + fromlist: tuple = (), level: int = 0) -> object: + if name == "truememory.ingest.transcript": + return fake + raise AssertionError("Stop count imported an unexpected module") + + namespace = {"Path": Path, "json": json, "__builtins__": dict(vars(builtins), __import__=safe_import)} + exec(compile(ast.Module(body=[node], type_ignores=[]), "synthetic-stop", "exec"), namespace) + return namespace["_has_enough_messages"], calls + + +def counted_reconciliation(parser: types.ModuleType, records: list[dict]) -> tuple[object, int]: + """Count key operations and binary-search reads in the actual outcome path.""" + operations = [0] + + class CountedText(str): + def __hash__(self) -> int: + operations[0] += 1 + return super().__hash__() + + def __eq__(self, other: object) -> bool: + operations[0] += 1 + return super().__eq__(other) + + decode = parser._decode_codex_record + + def counted_decode(entry: dict) -> object: + result = decode(entry) + if result is not None: + for key in ("projection", "phase", "turn_id", "item_id"): + value = getattr(result, key) + if isinstance(value, str): + setattr(result, key, CountedText(value)) + return result + + def counted_bisect(values: list[int], target: int) -> int: + class Positions: + def __len__(self) -> int: + return len(values) + + def __getitem__(self, index: int) -> int: + operations[0] += 1 + return values[index] + + return bisect_left(Positions(), target) + + with patch.object(parser, "_decode_codex_record", counted_decode), \ + patch.object(parser, "bisect_left", counted_bisect, create=True): + result = parser._parse_transcript_text_outcome(json.dumps(records)) + return result, operations[0] + + +class NativeRolloutTests(unittest.TestCase): + def setUp(self) -> None: + self.parser = load_parser() + + def check(self, records: list, expected: list[tuple[str, str]], status: str = "complete", + malformed: int = 0, unknown: int = 0) -> object: + for text in (json.dumps(records), "\n".join(json.dumps(record) for record in records)): + with patch.object(Path, "stat", side_effect=FileNotFoundError(errno.ENOENT, "synthetic missing")): + messages = self.parser.parse_transcript(text) + outcome = self.parser.parse_transcript_outcome(text) + self.assertEqual([(message.role, message.content) for message in messages], expected) + self.assertEqual(outcome.messages, messages) + self.assertEqual((outcome.status, outcome.total_records, outcome.malformed_records, + outcome.unrecognized_records), (status, len(records), malformed, unknown)) + self.assertNotIn(SENTINEL, repr(outcome)) + self.assertNotIn(TURN, repr(outcome)) + self.assertNotIn(SCAFFOLD, repr(outcome)) + return outcome + + def test_full_legacy_and_paginated_lifecycles(self) -> None: + for paginated in (False, True): + with self.subTest(paginated=paginated): + result = self.check(lifecycle(paginated=paginated), [("human", SENTINEL), ("assistant", "done")]) + self.assertEqual(self.parser.format_for_extraction(result.messages), + "User: " + SENTINEL + "\n\nAssistant: done") + + def test_original_blocks_and_exact_private_projection(self) -> None: + for parts in (("alpha", "beta"), (" a", "b "), ("", "alpha", "", "beta")): + for paginated in (False, True): + with self.subTest(parts=parts, paginated=paginated): + self.check(lifecycle(parts=parts, paginated=paginated), + [("human", SENTINEL), ("assistant", "\n".join(p for p in parts if p).strip())]) + records = lifecycle(parts=(" a", "b ")) + records[-1] = terminal("ab") + self.check(records, [("human", SENTINEL), ("assistant", "a\nb")], "partial", unknown=1) + records[-1] = terminal(" a\nb ") + self.check(records, [("human", SENTINEL), ("assistant", "a\nb")], "partial", unknown=1) + + def test_each_phase_and_blank_tail_preserve_summary(self) -> None: + for phase in (None, "commentary", "partial_answer", "final_answer"): + with self.subTest(phase=phase): + records = lifecycle(phase=phase) + records[-1:-1] = [event("agent_message", message=" \n"), response("assistant", (" \n",))] + self.check(records, [("human", SENTINEL), ("assistant", "done")]) + + def test_duplicate_occurrences_and_ids_are_not_deduplicated(self) -> None: + records = lifecycle() + records[-1:-1] = [event("agent_message", message="done"), response("assistant", ("done",))] + self.check(records, [("human", SENTINEL), ("assistant", "done"), ("assistant", "done")]) + self.check([response()] * 5, [("human", SENTINEL)] * 5) + + def test_extra_echo_cannot_borrow_terminal_credit(self) -> None: + for paginated in (False, True): + records = lifecycle(paginated=paginated) + records.insert(-1, copy.deepcopy(records[3])) + self.check(records, [("human", SENTINEL), ("assistant", "done")], "partial", unknown=1) + + def test_per_original_block_echoes_are_not_ordinary_finalization(self) -> None: + records = lifecycle(parts=("alpha", "beta")) + records[3:4] = [event("agent_message", message="alpha"), event("agent_message", message="beta")] + self.check(records, [("human", SENTINEL), ("assistant", "alpha\nbeta")], "partial", unknown=2) + + def test_missing_echoes_and_monotonic_occurrence_matching(self) -> None: + records = [start(), response("assistant", ("a",)), response("assistant", ("b",)), + event("agent_message", message="b"), terminal("b")] + self.check(records, [("assistant", "a"), ("assistant", "b")]) + records.insert(-1, event("agent_message", message="a")) + self.check(records, [("assistant", "a"), ("assistant", "b")], "partial", unknown=1) + + def test_summary_is_not_last_row_or_consumed_echo(self) -> None: + records = lifecycle(parts=("a",)) + records[-1:-1] = [response("assistant", ("b",)), event("stream_error", message=SENTINEL)] + self.check(records, [("human", SENTINEL), ("assistant", "a"), ("assistant", "b")]) + records[-1] = terminal(None) + self.check(records, [("human", SENTINEL), ("assistant", "a"), ("assistant", "b")]) + + def test_blank_echo_requires_canonical_occurrence(self) -> None: + self.check([start(), event("agent_message", message=" \n"), response("assistant", (" \n",)), terminal(None)], []) + self.check([start(), event("agent_message", message=" \n"), terminal(None)], [], "partial", unknown=1) + for invalid in ("", " \n", 1, [], {}): + with self.subTest(summary_type=type(invalid).__name__): + self.check([start(), terminal(invalid)], [], "partial", unknown=1) + + def test_rust_blank_predicate_is_separate_from_public_strip(self) -> None: + for nonblank in ("\u001c", "\u001d", "\u001e", "\u001f"): + self.check([start(), response("assistant", (nonblank,)), terminal(nonblank)], []) + for blank in ("\u0085", "\u00a0", "\u2003", "\u2028", "\u3000"): + self.check([start(), response("assistant", (blank,)), terminal(blank)], [], "partial", unknown=1) + + def test_lifecycle_aliases_and_missing_legacy_response_ids(self) -> None: + records = lifecycle() + records[0]["payload"]["type"] = "turn_started" + records[-1]["payload"]["type"] = "turn_complete" + for index in (1, 4): + records[index]["payload"].pop("internal_chat_message_metadata_passthrough") + records[index]["payload"].pop("id") + self.check(records, [("human", SENTINEL), ("assistant", "done")]) + + def test_orphan_terminal_requires_explicit_matching_scope(self) -> None: + self.check([response("assistant", ("done",)), terminal()], [("assistant", "done")]) + self.check([response("assistant", ("done",), turn=None), terminal()], + [("assistant", "done")], "partial", unknown=1) + self.check([terminal(), response("assistant", ("done",))], + [("assistant", "done")], "partial", unknown=1) + self.check([response("assistant", ("done",)), response("assistant", ("other",), turn="other"), terminal()], + [("assistant", "done"), ("assistant", "other")], "partial", unknown=1) + + def test_sessions_and_reused_turn_ids_are_occurrence_scopes(self) -> None: + self.check(lifecycle() + lifecycle(), [("human", SENTINEL), ("assistant", "done")] * 2) + self.check(lifecycle() + [{"type": "session_meta", "payload": {}}] + [start(), terminal()], + [("human", SENTINEL), ("assistant", "done")], "partial", unknown=1) + self.check(lifecycle() + [start("second"), terminal("done", "second")], + [("human", SENTINEL), ("assistant", "done")], "partial", unknown=1) + self.check([response("assistant", ("done",), turn=None), {"type": "session_meta", "payload": {}}, + event("agent_message", message="done")], [("assistant", "done")], "partial", unknown=1) + + def test_conflicting_turn_phase_and_item_identity_stay_partial(self) -> None: + for field, value in (("turn_id", "other-turn"), ("id", "other-item"), ("phase", "commentary")): + records = lifecycle(paginated=True, phase="final_answer") + target = records[3]["payload"] if field == "turn_id" else records[3]["payload"]["item"] + target[field] = value + result = self.parser._parse_transcript_text_outcome(json.dumps(records)) + self.assertEqual(result.status, "partial") + self.assertEqual(len(result.messages), 2) + records = lifecycle() + records[4]["payload"]["internal_chat_message_metadata_passthrough"]["turn_id"] = "other-turn" + self.check(records, [("human", SENTINEL), ("assistant", "done")], "partial", unknown=3) + + def test_duplicate_and_overlapping_terminals_do_not_manufacture_evidence(self) -> None: + records = lifecycle() + [terminal()] + self.check(records, [("human", SENTINEL), ("assistant", "done")], "partial", unknown=1) + records = [start(), response("assistant", ("done",)), start("other"), terminal("done", "other")] + result = self.parser._parse_transcript_text_outcome(json.dumps(records)) + self.assertEqual(result.status, "partial") + self.assertIn("codex_unsupported_lifecycle", result.error_categories) + + def test_events_cannot_explain_each_other_or_rewrites(self) -> None: + self.check([start(), event("agent_message", message="done"), terminal()], [], "partial", unknown=2) + records = lifecycle() + records[4] = response("assistant", ("original",)) + self.check(records, [("human", SENTINEL), ("assistant", "original")], "partial", unknown=2) + + def test_markup_and_mode_rewriting_are_explicit_coverage_gaps(self) -> None: + for parts in (("xhiddeny"), + ("before\n\nstep\n\nafter",)): + for mode in ("plan", "default", None): + records = lifecycle(parts=parts) + records[0]["payload"]["collaboration_mode_kind"] = mode + result = self.parser._parse_transcript_text_outcome(json.dumps(records)) + self.assertEqual(result.status, "partial") + self.assertIn("codex_unsupported_projection", result.error_categories) + self.assertEqual(len(result.messages), 2) + + def test_compaction_does_not_clear_original_evidence_or_emit_copies(self) -> None: + records = lifecycle() + records.insert(-1, {"type": "compacted", "payload": {"message": SCAFFOLD, + "replacement_history": [response("assistant", (SCAFFOLD,))["payload"]]}}) + self.check(records, [("human", SENTINEL), ("assistant", "done")]) + self.check([start(), records[-2], terminal(None)], []) + + def test_excluded_context_variants_do_not_unwrap_text(self) -> None: + top = ("session_meta", "turn_context", "compacted", "token_usage_record", "world_state", "retained_context", + "security_risk_score", "inter_agent_communication", "inter_agent_communication_metadata") + variants = ("reasoning", "agent_message", "function_call", "function_call_output", "custom_tool_call", + "custom_tool_call_output", "compaction", "compaction_summary", "configuration_update") + records = [{"type": kind, "payload": {"role": "user", "content": SENTINEL}} for kind in top] + records += [{"type": "response_item", "payload": {"type": kind, "role": "user", "content": SENTINEL}} + for kind in variants] + self.check(records, []) + + def test_outer_flags_are_strict_booleans(self) -> None: + for key in ("compaction_output", "inherited_user_message"): + for value in (0, 1, "false", "true", [], {}, None): + record = response() + record["metadata"] = {key: value} + with self.subTest(key=key, value=value): + self.check([record], [], "malformed", malformed=1) + record["metadata"] = {key: True} + self.check([record], []) + record["metadata"] = {key: False} + self.check([record], [("human", SENTINEL)]) + + def test_client_authored_does_not_override_scaffold(self) -> None: + record = annotate(response(), [annotation("harness")]) + record["metadata"] = {"client_authored": True} + self.check([record], []) + + def test_delivered_markers_exclude_text_and_unavailable_suffix(self) -> None: + record = response("assistant") + record["metadata"] = {"delivered_assistant_message": "codex:code-mode-delivery:v1:complete"} + self.check([record], []) + record["metadata"]["delivered_assistant_message"] = "codex:code-mode-delivery:v1:incomplete:" + SENTINEL + result = self.check([record], [], "partial", unknown=1) + self.assertIn("codex_delivery_unavailable", result.error_categories) + record["metadata"]["delivered_assistant_message"] = "other" + self.check([record], [], "partial", unknown=1) + record["metadata"]["delivered_assistant_message"] = None + self.check([record], [], "malformed", malformed=1) + self.check([response("assistant", ("codex:code-mode-delivery:v1:complete",))], + [("assistant", "codex:code-mode-delivery:v1:complete")]) + + def test_metadata_objects_and_array_alignment(self) -> None: + for bad in (1, "bad", [], False): + for outer in (True, False): + record = response() + if outer: + record["metadata"] = bad + else: + record["payload"]["internal_chat_message_metadata_passthrough"] = bad + self.check([record], [], "malformed", malformed=1) + for key, bad in (("content_item_metadata", []), ("content_item_metadata", [None]), + ("content_item_metadata", [{}, {}]), ("content_item_metadata", {}), + ("content_item_kinds", [None]), ("content_item_kinds", []), ("content_item_kinds", "unknown")): + record = response() + record["payload"]["internal_chat_message_metadata_passthrough"][key] = bad + self.check([record], [], "malformed", malformed=1) + self.check([annotate(response(), None, None)], [("human", SENTINEL)]) + + def test_per_index_provenance_does_not_shift_after_media(self) -> None: + record = response(parts=(SCAFFOLD, SENTINEL, SCAFFOLD)) + record["payload"]["content"][0] = {"type": "input_image", "image_url": "synthetic-image"} + annotate(record, [annotation("harness"), annotation("user"), annotation("agents_md")]) + result = self.check([record], [("human", SENTINEL)]) + self.assertNotIn(SCAFFOLD, self.parser.format_for_extraction(result.messages)) + record["payload"]["internal_chat_message_metadata_passthrough"]["content_item_metadata"][0] = annotation("user") + self.check([record], [("human", SENTINEL)], "partial", unknown=1) + + def test_provenance_and_legacy_fields(self) -> None: + exclusions = [annotation("developer_instructions", from_additional_requirements=False), annotation("skills"), + annotation("agents_md"), annotation("harness", hook=True), annotation("tool", namespace="functions"), + {"harness_injected": True}, {"source_tool_namespace": "functions", "harness_injected": False}] + for attribution in exclusions: + self.check([annotate(response(), [attribution])], []) + for attribution in ({}, {"harness_injected": False}, {"source_tool_namespace": None}, annotation("user")): + self.check([annotate(response(), [attribution])], [("human", SENTINEL)]) + for kind in ("unknown", "user_configured", "future"): + self.check([annotate(response(), [annotation(kind)])], [], "partial", unknown=1) + + def test_nested_metadata_types_never_fall_back_to_user(self) -> None: + values = [{"provenance": None}, {"provenance": "user"}, {"provenance": {"type": []}}, + annotation("harness", hook=1), annotation("developer_instructions"), + annotation("developer_instructions", from_additional_requirements="false"), + annotation("tool", namespace=1), {"harness_injected": 1}, {"source_tool_namespace": []}] + for attribution in values: + self.check([annotate(response(), [attribution])], [], "malformed", malformed=1) + + def test_kinds_fallback_conflict_and_lookalike_prose(self) -> None: + for kind in ("agents_md.instructions", "hooks.additional_context", "guardian.retained_instructions"): + self.check([annotate(response(), None, [kind])], []) + self.check([annotate(response(), [annotation("user")], [kind])], [], "partial", unknown=1) + self.check([annotate(response(), None, ["user.future"])], [], "partial", unknown=1) + self.check([annotate(response(), [annotation("harness")], ["future.fragment"])], []) + self.check([response(parts=("# AGENTS.md ordinary quoted text",))], [("human", "# AGENTS.md ordinary quoted text")]) + for kind in ("", "unknown"): + self.check([annotate(response(), None, [kind])], [("human", SENTINEL)]) + + def test_excluded_equal_text_cannot_supply_echo_or_terminal_credit(self) -> None: + excluded = [response("assistant", ("done",)) for _ in range(5)] + excluded[0]["metadata"] = {"inherited_user_message": True} + excluded[1]["metadata"] = {"compaction_output": True} + excluded[2]["metadata"] = {"delivered_assistant_message": "codex:code-mode-delivery:v1:complete"} + annotate(excluded[3], [annotation("harness")]) + annotate(excluded[4], [annotation("unknown")]) + for index, record in enumerate(excluded): + self.check([start(), event("agent_message", message="done"), record, terminal()], [], + "partial", unknown=3 if index == 4 else 2) + + def test_record_level_counts_and_private_diagnostics(self) -> None: + record = response() + record["payload"]["content"] = [{"type": "future", "text": SENTINEL}, {"type": "input_image"}, + {"type": "input_text", "text": 3}, 5, + {"type": "input_text", "text": SENTINEL}] + record["timestamp"] = {SENTINEL: "bad"} + result = self.check([record], [("human", SENTINEL)], "partial", malformed=1, unknown=1) + self.assertEqual(result.messages[0].timestamp, "") + self.assertEqual(result.total_records, 1) + + def test_unknown_native_and_events_are_not_empty_complete(self) -> None: + for record in ({"type": "future", "payload": {"content": SENTINEL}}, + {"type": "realtime_item", "payload": {"text": SENTINEL}}, + {"type": "response_item", "payload": {"type": "future", "content": SENTINEL}}, + event("future", text=SENTINEL), event("user_message", message=SENTINEL), + event("item_completed", item={"type": "Future", "text": SENTINEL})): + self.check([record], [], "partial", unknown=1) + for record in ({"type": "response_item", "payload": []}, + {"type": "response_item", "payload": {"type": "message", "role": "user", "content": SENTINEL}}, + {"type": "event_msg", "payload": {"type": []}}): + self.check([record], [], "malformed", malformed=1) + for invalid in ([], {}, 1, None): + self.check([{"type": invalid, "payload": {}}], [], "malformed", malformed=1) + + def test_user_output_text_is_not_a_normal_user_projection(self) -> None: + record = response() + record["payload"]["content"][0]["type"] = "output_text" + self.check([record], [("human", SENTINEL)]) + self.check([record, event("user_message", message=SENTINEL)], [("human", SENTINEL)], "partial", unknown=1) + + def test_event_media_and_unknown_typed_items_stay_partial(self) -> None: + for key in ("images", "audio", "local_images", "file_ids"): + records = lifecycle() + records[2]["payload"][key] = [SENTINEL] + self.check(records, [("human", SENTINEL), ("assistant", "done")], "partial", unknown=1) + records = lifecycle(paginated=True) + records[3]["payload"]["item"]["content"].append({"type": "Text", "text": "extra"}) + self.check(records, [("human", SENTINEL), ("assistant", "done")], "partial", unknown=1) + + def test_claude_and_generic_compatibility(self) -> None: + records = [{"type": "user", "message": {"role": "user", "content": "hello"}}, + {"type": "assistant", "message": {"content": [{"type": "thinking", "thinking": SCAFFOLD}, + {"type": "text", "text": "answer"}]}}, + {"type": "user", "message": {"content": [{"type": "tool_result", "content": "tool"}]}}, + {"role": "user", "content": "generic"}, {"type": "progress"}] + result = self.check(records, [("human", "hello"), ("assistant", "answer"), ("tool_result", "tool"), ("human", "generic")]) + self.assertEqual(self.parser.format_for_extraction(result.messages), "User: hello\n\nAssistant: answer\n\nUser: generic") + + def test_generic_records_keep_incidental_payload_fields(self) -> None: + for payload in (None, {}, {"type": "message", "role": "assistant", "content": SCAFFOLD}): + for type_fields in ({}, {"type": None}, {"type": ""}): + records = [{"role": role, "content": text, "payload": payload, **type_fields} + for role, text in (("user", SENTINEL), ("assistant", "answer"))] + self.check(records, [("human", SENTINEL), ("assistant", "answer")]) + self.check([{"type": "future_native", "payload": {}, "role": "user", "content": SENTINEL}], + [], "partial", unknown=1) + self.check([{"role": "observer", "content": SENTINEL, "payload": None}], [("observer", SENTINEL)]) + + def test_index_preserves_absent_and_explicit_identity_compatibility(self) -> None: + first = response("assistant", ("same",), phase="commentary") + first["payload"].pop("id") + second = response("assistant", ("same",), phase="final_answer", item="second") + third = response("assistant", ("same",), phase="final_answer", item="third") + records = [start(), first, second, third, + completed("assistant", "same", item="second", phase="commentary"), + event("agent_message", message="same", phase="final_answer"), + completed("assistant", "same", item="third"), terminal("same")] + self.check(records, [("assistant", "same")] * 3) + records.insert(-1, completed("assistant", "same", item="third")) + self.check(records, [("assistant", "same")] * 3, "partial", unknown=1) + + def test_legacy_exclusions_keep_incidental_payload_fields(self) -> None: + for kind in sorted(self.parser._NON_CONVERSATION_TYPES): + for selector in ({"type": kind}, {"role": kind}, {"type": None, "role": kind}): + for extra in ({}, {"payload": None}, {"payload": {}}, + {"payload": {"type": "message", "role": "assistant", "content": SCAFFOLD}}): + with self.subTest(kind=kind, selector=selector, payload_present="payload" in extra): + self.check([{**selector, **extra}], []) + self.check([{"type": "future_native", "role": "summary", "payload": {}}], + [], "partial", unknown=1) + self.check([{"type": "response_item", "role": "summary", "payload": {}}], + [], "malformed", malformed=1) + + def test_identity_filters_do_not_rewind_role_cursors(self) -> None: + records = [response("assistant", ("same",), turn="first", item="first"), + response(parts=("same",), turn="second"), + response("assistant", ("same",), turn="second", item="second"), + completed("assistant", "same", turn="second", item="second"), + event("user_message", message="same"), + completed("assistant", "same", turn="first", item="first")] + self.check(records, [("assistant", "same"), ("human", "same"), ("assistant", "same")], "partial", unknown=1) + + def test_unmatched_echo_work_is_bounded_including_metadata_misses(self) -> None: + for mismatch in ("text", "phase", "item", "turn"): + previous = None + for size in (100, 200, 400): + with self.subTest(mismatch=mismatch, size=size): + originals = [response("assistant", ("same",), phase="commentary") for _ in range(size)] + if mismatch == "text": + echo = event("agent_message", message="different") + elif mismatch == "phase": + echo = event("agent_message", message="same", phase="final_answer") + elif mismatch == "item": + echo = completed("assistant", "same", item="different") + else: + echo = completed("assistant", "same", turn="different") + result, operations = counted_reconciliation(self.parser, originals + [echo] * size) + self.assertEqual((len(result.messages), result.unrecognized_records), (size, size)) + self.assertEqual(result.status, "partial") + # Eight entries per response; at most two indexed event + # queries. Allow linear key work plus binary-search reads. + self.assertLessEqual(operations, 96 * size + 4 * size * (size.bit_length() + 1)) + if previous is not None: + self.assertLessEqual(operations, 3 * previous + 32) + previous = operations + + def test_syntax_accounting_is_not_replaced_by_native_counts(self) -> None: + text = json.dumps(response()) + "\n{bad\n" + json.dumps(event("future")) + result = self.parser._parse_transcript_text_outcome(text) + self.assertEqual((result.status, result.total_records, result.malformed_records, result.unrecognized_records), + ("partial", 3, 1, 1)) + self.assertEqual(self.parser._parse_transcript_text_outcome("[bad").status, "salvaged") + + +class StopCountTests(unittest.TestCase): + def setUp(self) -> None: + self.parser = load_parser() + + def check(self, text: str, expected: bool, threshold: int = 5) -> None: + count, calls = load_stop_count(self.parser, text) + with patch.object(Path, "read_text", side_effect=AssertionError("Stop must not read directly")), \ + patch.object(Path, "stat", side_effect=AssertionError("Stop must not stat directly")): + self.assertIs(count("synthetic-transcript.jsonl", threshold), expected) + self.assertEqual(calls, [Path("synthetic-transcript.jsonl")]) + + def test_four_original_native_failures(self) -> None: + five = [response()] * 5 + self.check(json.dumps(five), True) + self.check("\n".join(json.dumps(item) for item in five), True) + self.check(json.dumps(response(parts=(SENTINEL * 40,))), False) + self.check(json.dumps({"type": "session_meta", "payload": {"metadata": SENTINEL * 40}}), False) + + def test_events_and_scaffolding_do_not_inflate_turns(self) -> None: + self.check(json.dumps([response()] + [event("user_message", message=SENTINEL)] * 5), False) + self.check(json.dumps([annotate(response(), [annotation("harness")])] * 5), False) + self.check(json.dumps([response(parts=("synthetic",))] * 5), False) + self.check(json.dumps([response(parts=(" \n",))] * 5), False) + self.check(json.dumps([response(parts=("one", "two", "three", "four", "five"))]), False) + + def test_actual_threshold_duplicates_and_partial_supported_turns(self) -> None: + self.check(json.dumps([response()] * 4), False) + self.check(json.dumps([response()] * 5), True) + self.check(json.dumps([response()] * 5 + [event("future")]), True) + self.check(json.dumps([response()] * 5), False, threshold=6) + self.check(json.dumps(lifecycle(paginated=True)), True, threshold=1) + + def test_claude_tools_plain_text_and_directives(self) -> None: + self.check(json.dumps([{"type": "user", "message": {"content": [{"type": "tool_result", "content": "result"}]}}] * 5), False) + self.check("Human: one\nAssistant: reply\nUser: two", True, threshold=2) + self.check("ordinary unmarked text " * 100, False) + self.check(json.dumps([response(parts=("Use the synthetic short format",))] * 5), True) + + def test_unreadable_is_false_even_at_zero_threshold(self) -> None: + with patch.object(self.parser, "_parse_transcript_text_outcome", return_value=self.parser.TranscriptOutcome([], "unreadable")): + self.check("", False, threshold=0) + + +class InMemoryPathTests(unittest.TestCase): + def test_native_explicit_file_path_preserves_version_and_privacy(self) -> None: + parser = load_parser() + data = ("\n".join(json.dumps(record) for record in lifecycle())).encode("utf-8") + info = types.SimpleNamespace(st_dev=1, st_ino=2, st_size=len(data), st_mtime_ns=3, st_ctime_ns=4) + + class Reader(io.BytesIO): + def fileno(self) -> int: + return 41 + + with patch.object(Path, "open", side_effect=lambda *args, **kwargs: Reader(data)), \ + patch.object(parser.os, "fstat", return_value=info), \ + patch.object(Path, "stat", side_effect=AssertionError("Explicit path must not probe")): + result = parser.parse_transcript_outcome(Path("synthetic-private-path.jsonl")) + self.assertTrue(result.complete) + self.assertTrue(result.file_version.stable) + self.assertEqual(result.total_records, 6) + self.assertEqual(len(result.messages), 2) + self.assertNotIn("synthetic-private-path", repr(result)) + + +class MutationControlTests(unittest.TestCase): + """The witnesses must reject the specific earlier unsafe implementations.""" + + def mutate(self, before: str, after: str) -> types.ModuleType: + source = (ROOT / "truememory/ingest/transcript.py").read_text(encoding="utf-8") + self.assertEqual(source.count(before), 1) + return load_parser(source.replace(before, after)) + + def outcome(self, parser: types.ModuleType, records: list[dict]) -> object: + return parser._parse_transcript_text_outcome(json.dumps(records)) + + def test_newline_or_trimmed_projection_breaks_ordinary_lifecycle(self) -> None: + for replacement in ('raw = "\\n".join(parts)', 'raw = "".join(parts).strip()'): + with self.subTest(replacement=replacement): + parser = self.mutate('raw = "".join(parts)', replacement) + self.assertEqual(self.outcome(parser, lifecycle(parts=(" a", "b "))).status, "partial") + + def test_removing_provenance_filter_leaks_scaffold(self) -> None: + parser = self.mutate('if attribution == "excluded":', 'if False:') + records = [annotate(response(parts=(SCAFFOLD,)), [annotation("harness")])] + self.assertEqual([message.content for message in self.outcome(parser, records).messages], [SCAFFOLD]) + self.assertEqual(self.outcome(load_parser(), records).messages, []) + + def test_ignoring_event_reconciliation_falsely_certifies_event_only(self) -> None: + parser = self.mutate('_reconcile_codex(native_records)', 'pass # synthetic mutation') + records = [start(), event("agent_message", message=SENTINEL), terminal(SENTINEL)] + self.assertTrue(self.outcome(parser, records).complete) + self.assertEqual(self.outcome(load_parser(), records).unrecognized_records, 2) + + def test_last_raw_row_only_rule_loses_valid_terminal(self) -> None: + parser = self.mutate('for before, candidate in reversed(candidates)', + 'for before, candidate in candidates[-1:]') + records = lifecycle(parts=("a",)) + records.insert(-1, response("assistant", ("b",))) + self.assertEqual(self.outcome(parser, records).status, "partial") + self.assertTrue(self.outcome(load_parser(), records).complete) + + def test_echoes_must_not_match_only_preceding_responses(self) -> None: + source = (ROOT / "truememory/ingest/transcript.py").read_text(encoding="utf-8") + source = source.replace('for _, event in items:', 'for event_index, event in items:') + target = 'position = _codex_next_echo(credit_index, event, cursors.get(event.role, 0))' + self.assertEqual(source.count(target), 1) + source = source.replace(target, target + '\n if position is not None and credits[position][0] >= event_index:\n position = None') + parser = load_parser(source) + self.assertEqual(self.outcome(parser, lifecycle()).status, "partial") + self.assertTrue(self.outcome(load_parser(), lifecycle()).complete) + + def test_terminal_must_not_require_final_phase(self) -> None: + parser = self.mutate('candidate.projection == terminal.projection and not candidate.malformed', + 'candidate.phase == "final_answer" and candidate.projection == terminal.projection and not candidate.malformed') + self.assertEqual(self.outcome(parser, lifecycle(phase="commentary")).status, "partial") + self.assertTrue(self.outcome(load_parser(), lifecycle(phase="commentary")).complete) + + def test_session_reset_is_necessary_even_without_turn_boundaries(self) -> None: + parser = self.mutate('if record.kind == "session_meta":', 'if False:') + records = [response("assistant", ("done",), turn=None), {"type": "session_meta", "payload": {}}, + event("agent_message", message="done")] + self.assertTrue(self.outcome(parser, records).complete) + self.assertEqual(self.outcome(load_parser(), records).status, "partial") + + def test_terminal_and_echo_credit_must_remain_independent(self) -> None: + parser = self.mutate('cursors[event.role] = position + 1', + 'cursors[event.role] = position + 1\n credits[position][1].projection = None') + self.assertEqual(self.outcome(parser, lifecycle()).status, "partial") + self.assertTrue(self.outcome(load_parser(), lifecycle()).complete) + + def test_echo_credit_must_be_consumed_once(self) -> None: + parser = self.mutate('cursors[event.role] = position + 1', 'cursors[event.role] = 0') + records = lifecycle() + records.insert(-1, event("agent_message", message="done")) + self.assertTrue(self.outcome(parser, records).complete) + self.assertEqual(self.outcome(load_parser(), records).status, "partial") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_502_config.py b/tests/test_issue_502_config.py index 9d03caf0..5bb1fa53 100644 --- a/tests/test_issue_502_config.py +++ b/tests/test_issue_502_config.py @@ -16,6 +16,7 @@ import json import os +from pathlib import Path import threading import time @@ -86,11 +87,14 @@ def test_load_config_returns_cached_copy(self, server, monkeypatch): # Second call within staleness window, file NOT modified on disk # (mtime unchanged) — should use cache, no disk read read_count = 0 - original_read_text = type(server._CONFIG_PATH).read_text + config_path = server._CONFIG_PATH + invoking_thread = threading.get_ident() + original_read_text = type(config_path).read_text - def counting_read(self_path, *args, **kwargs): + def counting_read(self_path: Path, *args: object, **kwargs: object) -> str: nonlocal read_count - read_count += 1 + if self_path == config_path and threading.get_ident() == invoking_thread: + read_count += 1 return original_read_text(self_path, *args, **kwargs) monkeypatch.setattr(type(server._CONFIG_PATH), "read_text", counting_read) @@ -99,6 +103,28 @@ def counting_read(self_path, *args, **kwargs): assert result1 == result2, ( "_load_config() did not return cached result within staleness window" ) + + # Prove that unrelated reads cannot contaminate this caller's count. + # Run controls after the second call so they cannot consume its TTL. + unrelated_path = config_path.with_name("unrelated.json") + unrelated_path.write_text(json.dumps(cfg), encoding="utf-8") + assert json.loads(unrelated_path.read_text(encoding="utf-8")) == cfg + background_results: list[str] = [] + background_errors: list[Exception] = [] + + def background_read() -> None: + try: + background_results.append(config_path.read_text(encoding="utf-8")) + except Exception as error: + background_errors.append(error) + + reader = threading.Thread(target=background_read) + reader.start() + reader.join(timeout=5) + assert not reader.is_alive() + assert not background_errors + assert len(background_results) == 1 + assert json.loads(background_results[0]) == cfg assert read_count == 0, ( f"_load_config() re-read disk {read_count} times within staleness window" ) diff --git a/tests/test_platform_compat.py b/tests/test_platform_compat.py index 6cdfcd24..de48b509 100644 --- a/tests/test_platform_compat.py +++ b/tests/test_platform_compat.py @@ -85,6 +85,7 @@ def _tmp_config_dir(tmp_path, monkeypatch, ms): (home / ".truememory").mkdir() monkeypatch.setattr(ms, "_TRUEMEMORY_DIR", home / ".truememory") monkeypatch.setattr(ms, "_CONFIG_PATH", home / ".truememory" / "config.json") + monkeypatch.setattr(ms, "_CONFIG_LOCK_PATH", home / ".truememory" / "config.json.lock") return home @@ -158,24 +159,36 @@ def test_ingest_cli_save_warns_on_windows_when_api_key_present( tmp_path, monkeypatch, capsys ): from truememory.ingest import cli as ic - cfg_path = tmp_path / ".truememory" / "config.json" + import truememory.mcp_server as ms + home = _tmp_config_dir(tmp_path, monkeypatch, ms) + cfg_path = home / ".truememory" / "config.json" monkeypatch.setattr(ic, "_TRUEMEMORY_CONFIG_PATH", cfg_path) monkeypatch.setattr(ic.sys, "platform", "win32") + assert ic._TRUEMEMORY_CONFIG_PATH == ms._CONFIG_PATH == cfg_path + assert ms._TRUEMEMORY_DIR == cfg_path.parent + assert ms._CONFIG_LOCK_PATH == cfg_path.parent / "config.json.lock" ic._save_truememory_config({"tier": "pro", "openrouter_api_key": "sk-or-fake"}) captured = capsys.readouterr() assert "windows" in captured.err.lower() assert "environment variable" in captured.err.lower() + assert json.loads(cfg_path.read_text(encoding="utf-8")) == {"tier": "pro", "openrouter_api_key": "sk-or-fake"} def test_ingest_cli_save_silent_on_posix(tmp_path, monkeypatch, capsys): from truememory.ingest import cli as ic - cfg_path = tmp_path / ".truememory" / "config.json" + import truememory.mcp_server as ms + home = _tmp_config_dir(tmp_path, monkeypatch, ms) + cfg_path = home / ".truememory" / "config.json" monkeypatch.setattr(ic, "_TRUEMEMORY_CONFIG_PATH", cfg_path) monkeypatch.setattr(ic.sys, "platform", "linux") + assert ic._TRUEMEMORY_CONFIG_PATH == ms._CONFIG_PATH == cfg_path + assert ms._TRUEMEMORY_DIR == cfg_path.parent + assert ms._CONFIG_LOCK_PATH == cfg_path.parent / "config.json.lock" ic._save_truememory_config({"tier": "pro", "openrouter_api_key": "sk-or-fake"}) captured = capsys.readouterr() assert captured.err == "" + assert json.loads(cfg_path.read_text(encoding="utf-8")) == {"tier": "pro", "openrouter_api_key": "sk-or-fake"} diff --git a/truememory/ingest/hooks/stop.py b/truememory/ingest/hooks/stop.py index 26766d1a..d8e75083 100755 --- a/truememory/ingest/hooks/stop.py +++ b/truememory/ingest/hooks/stop.py @@ -224,56 +224,15 @@ def _writable_dirs_ok() -> bool: def _has_enough_messages(transcript_path: str, min_messages: int) -> bool: - """Check whether the transcript has at least `min_messages` user turns. + """Count meaningful canonical human turns, independently of completion coverage.""" + from truememory.ingest.transcript import parse_transcript_outcome, strip_truememory_blocks - Parses the transcript properly instead of substring-counting so that - conversations containing the literal strings "human"/"user" in content - don't inflate the count. - """ - try: - content = Path(transcript_path).read_text(encoding="utf-8", errors="replace") - except Exception: - return False - - if not content.strip(): + outcome = parse_transcript_outcome(Path(transcript_path)) + if outcome.status == "unreadable": return False - - # Try JSON array first (Claude Code format) - count = 0 - try: - if content.lstrip().startswith("["): - data = json.loads(content) - if isinstance(data, list): - for entry in data: - if isinstance(entry, dict): - role = entry.get("type") or entry.get("role") or "" - if role in ("human", "user"): - count += 1 - return count >= min_messages - except json.JSONDecodeError: - pass - - # Try JSONL - try: - for line in content.splitlines(): - line = line.strip() - if not line: - continue - try: - entry = json.loads(line) - if isinstance(entry, dict): - role = entry.get("type") or entry.get("role") or "" - if role in ("human", "user"): - count += 1 - except json.JSONDecodeError: - continue - if count > 0: - return count >= min_messages - except Exception: - pass - - # Fall back to length heuristic for plain text - return len(content) > min_messages * 50 + count = sum(message.role == "human" and bool(strip_truememory_blocks(message.content).strip()) + for message in outcome.messages) + return count >= min_messages def _count_active_ingest_processes() -> int: diff --git a/truememory/ingest/transcript.py b/truememory/ingest/transcript.py index b0fba5ff..64a40f47 100644 --- a/truememory/ingest/transcript.py +++ b/truememory/ingest/transcript.py @@ -21,6 +21,7 @@ import os import re import stat +from bisect import bisect_left from dataclasses import dataclass, field from pathlib import Path @@ -85,6 +86,425 @@ def complete(self) -> bool: return self.status == "complete" +# Pinned Codex rollout grammar: openai/codex 0b755b1945bf4a31560df2ce1469aeb044dccbc9. +# These are exact exclusions, never prefix rules or recursive payload discovery. +_CODEX_CONTEXT = { + "session_meta", "turn_context", "compacted", "token_usage_record", "world_state", + "retained_context", "security_risk_score", "inter_agent_communication", + "inter_agent_communication_metadata", +} +_CODEX_RESPONSE_CONTEXT = { + "additional_tools", "agent_message", "reasoning", "local_shell_call", "function_call", + "function_call_output", "custom_tool_call", "custom_tool_call_output", "tool_search_call", + "tool_search_output", "web_search_call", "image_generation_call", "compaction", + "compaction_summary", "configuration_update", "compaction_trigger", "context_compaction", +} +_CODEX_EVENT_CONTEXT = { + "token_count", "agent_reasoning", "agent_reasoning_raw_content", "context_compacted", + "exec_command_begin", "exec_command_end", "exec_command_output_delta", + "mcp_tool_call_begin", "mcp_tool_call_end", "dynamic_tool_call_request", + "dynamic_tool_call_response", "patch_apply_begin", "patch_apply_end", "patch_apply_updated", + "web_search_begin", "web_search_end", "view_image_tool_call", "image_generation_begin", + "image_generation_end", "stream_error", "stream_info", "error", "warning", +} +_CODEX_ITEM_CONTEXT = { + "FunctionCallOutput", "HookPrompt", "Reasoning", "CommandExecution", "DynamicToolCall", + "CollabAgentToolCall", "SubAgentActivity", "WebSearch", "ImageView", "ImageGeneration", + "EnteredReviewMode", "ExitedReviewMode", "FileChange", "McpToolCall", "ContextCompaction", +} +_CODEX_SCAFFOLD_KINDS = { + "agents_md.instructions", "hooks.additional_context", "guardian.retained_instructions", +} +_CODEX_PHASES = {"commentary", "partial_answer", "final_answer"} +_CODEX_START = {"task_started", "turn_started"} +_CODEX_END = {"task_complete", "turn_complete"} +_CODEX_ANY = object() +_CODEX_WHITE_SPACE = "\t\n\v\f\r \u0085\u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000" + + +@dataclass +class _CodexRecord: + kind: str + message: Message | None = field(default=None, repr=False) + role: str = "" + projection: str | None = field(default=None, repr=False) + turn_id: str | None = field(default=None, repr=False) + item_id: str | None = field(default=None, repr=False) + phase: str | None = None + payload: dict = field(default_factory=dict, repr=False) + malformed: bool = False + unrecognized: bool = False + errors: set[str] = field(default_factory=set) + + def reject(self, category: str, *, malformed: bool = False) -> None: + self.errors.add(category) + if malformed: + self.malformed = True + else: + self.unrecognized = True + + +def _codex_attribution(value: dict) -> tuple[str, bool]: + """Return absent/user/excluded/unknown, validating even legacy selectors.""" + for key, expected in (("harness_injected", bool), ("source_tool_namespace", str)): + if key in value and value[key] is not None and type(value[key]) is not expected: + return "unknown", True + if "provenance" in value: + provenance = value["provenance"] + if not isinstance(provenance, dict) or not isinstance(provenance.get("type"), str): + return "unknown", True + kind = provenance["type"] + if kind == "developer_instructions": + if type(provenance.get("from_additional_requirements")) is not bool: + return "unknown", True + if kind == "harness" and type(provenance.get("hook", False)) is not bool: + return "unknown", True + if kind == "tool" and provenance.get("namespace") is not None: + if not isinstance(provenance["namespace"], str): + return "unknown", True + if kind == "user": + return "user", False + if kind in {"developer_instructions", "skills", "agents_md", "harness", "tool"}: + return "excluded", False + return "unknown", False + if value.get("source_tool_namespace") is not None or value.get("harness_injected") is True: + return "excluded", False + # Empty historical metadata and harness_injected:false are not positive + # provenance. They use only the documented older role/text compatibility. + return "absent", False + + +def _codex_optional_string(record: _CodexRecord, payload: dict, key: str) -> str | None: + value = payload.get(key) + if value is not None and not isinstance(value, str): + record.reject("codex_invalid_metadata", malformed=True) + return None + return value or None + + +def _decode_codex_record(entry: dict) -> _CodexRecord | None: + """Decode exact native envelopes without interpreting arbitrary nested prose.""" + kind = entry.get("type") + native = kind in _CODEX_CONTEXT if isinstance(kind, str) else False + native |= kind in ("response_item", "event_msg", "realtime_item") + if not native: + generic_kind = kind or entry.get("role") + if ("payload" not in entry + or (isinstance(generic_kind, str) and generic_kind in _NON_CONVERSATION_TYPES) + or generic_kind in ("user", "human", "assistant", "tool_use", "tool_result", "system") + or (not kind and "role" in entry and "content" in entry)): + return None + record = _CodexRecord(kind=kind if isinstance(kind, str) else "unknown") + if not isinstance(kind, str): + record.reject("codex_invalid_record", malformed=True) + return record + if not isinstance(entry.get("timestamp", ""), str): + record.reject("codex_invalid_metadata", malformed=True) + payload = entry.get("payload") + if not isinstance(payload, dict): + record.reject("codex_invalid_record", malformed=True) + return record + record.payload = payload + if kind in _CODEX_CONTEXT: + return record + if kind == "event_msg": + return _decode_codex_event(record) + if kind != "response_item": + record.reject("codex_unrecognized_record") + return record + variant = payload.get("type") + if not isinstance(variant, str): + record.reject("codex_invalid_record", malformed=True) + return record + if variant in _CODEX_RESPONSE_CONTEXT: + return record + if variant != "message": + record.reject("codex_unrecognized_record") + return record + role, content = payload.get("role"), payload.get("content") + if not isinstance(role, str) or not isinstance(content, list): + record.reject("codex_invalid_record", malformed=True) + return record + if role in ("system", "developer", "tool"): + return record + if role not in ("user", "assistant"): + record.reject("codex_unrecognized_record") + return record + record.role = "human" if role == "user" else "assistant" + outer, metadata = entry.get("metadata"), payload.get("internal_chat_message_metadata_passthrough") + if any(value is not None and not isinstance(value, dict) for value in (outer, metadata)): + record.reject("codex_invalid_metadata", malformed=True) + return record + outer, metadata = outer or {}, metadata or {} + excluded = False + for key in ("compaction_output", "inherited_user_message"): + if key in outer: + if type(outer[key]) is not bool: + record.reject("codex_invalid_metadata", malformed=True) + return record + excluded |= outer[key] + if "delivered_assistant_message" in outer: + marker = outer["delivered_assistant_message"] + if not isinstance(marker, str): + record.reject("codex_invalid_metadata", malformed=True) + elif marker.startswith("codex:code-mode-delivery:v1:incomplete:"): + record.reject("codex_delivery_unavailable") + elif marker != "codex:code-mode-delivery:v1:complete": + record.reject("codex_unrecognized_record") + return record + if excluded: + return record + record.turn_id = _codex_optional_string(record, metadata, "turn_id") + record.item_id = _codex_optional_string(record, payload, "id") + record.phase = _codex_optional_string(record, payload, "phase") + if record.phase is not None and record.phase not in _CODEX_PHASES: + record.reject("codex_unrecognized_record") + annotations, kinds = metadata.get("content_item_metadata"), metadata.get("content_item_kinds") + for values, value_type in ((annotations, dict), (kinds, str)): + if values is not None and (not isinstance(values, list) or len(values) != len(content) + or any(not isinstance(value, value_type) for value in values)): + record.reject("codex_invalid_metadata", malformed=True) + return record + parts: list[str] = [] + projection_allowed = True + for index, block in enumerate(content): + attribution, bad = _codex_attribution(annotations[index]) if annotations is not None else ("absent", False) + block_kind = kinds[index] if kinds is not None else "" + if bad: + record.reject("codex_invalid_metadata", malformed=True) + projection_allowed = False + continue + if attribution == "excluded": + # Proven attribution already covers generated context, including + # new fragment kinds. Kind fallback cannot re-authorize it. + projection_allowed = False + continue + if block_kind in _CODEX_SCAFFOLD_KINDS: + if attribution == "user": + record.reject("codex_unrecognized_provenance") + projection_allowed = False + continue + if block_kind not in ("", "unknown"): + record.reject("codex_unrecognized_provenance") + projection_allowed = False + continue + if attribution == "unknown" or (attribution == "user" and role != "user"): + record.reject("codex_unrecognized_provenance") + projection_allowed = False + continue + if not isinstance(block, dict) or not isinstance(block.get("type"), str): + record.reject("codex_invalid_content", malformed=True) + continue + if block["type"] not in ("input_text", "output_text"): + record.reject("codex_unrecognized_content") + continue + if not isinstance(block.get("text"), str): + record.reject("codex_invalid_content", malformed=True) + continue + parts.append(block["text"]) + if role == "user" and block["type"] != "input_text": + projection_allowed = False + public_text = "\n".join(part for part in parts if part).strip() + if public_text: + timestamp = entry.get("timestamp", "") + record.message = Message(record.role, public_text, timestamp if isinstance(timestamp, str) else "") + if projection_allowed and not record.malformed and not record.unrecognized: + raw = "".join(parts) + # Contributor/citation/plan rewriting is outside this plain-text + # projection. Never try alternative normalizations until an echo fits. + if role == "assistant" and any(tag in raw for tag in ("", "", "")): + record.reject("codex_unsupported_projection") + else: + record.projection = raw + return record + + +def _decode_codex_event(record: _CodexRecord) -> _CodexRecord: + payload = record.payload + kind = payload.get("type") + if not isinstance(kind, str): + record.reject("codex_invalid_record", malformed=True) + return record + record.kind = kind + if kind in _CODEX_START | _CODEX_END | {"turn_aborted"}: + record.turn_id = _codex_optional_string(record, payload, "turn_id") + if record.turn_id is None: + record.reject("codex_unsupported_lifecycle") + if kind in _CODEX_END: + summary = payload.get("last_agent_message") + if summary is not None: + if not isinstance(summary, str) or not summary.strip(_CODEX_WHITE_SPACE): + record.reject("codex_unmatched_terminal") + else: + record.projection = summary + return record + if kind in _CODEX_EVENT_CONTEXT: + return record + if kind in ("item_started", "item_completed"): + item = payload.get("item") + if not isinstance(item, dict) or not isinstance(item.get("type"), str): + record.reject("codex_invalid_record", malformed=True) + return record + item_kind = item["type"] + if item_kind in _CODEX_ITEM_CONTEXT: + return record + if item_kind not in ("UserMessage", "AgentMessage"): + record.reject("codex_unrecognized_record") + return record + if kind == "item_started": + return record + record.turn_id = _codex_optional_string(record, payload, "turn_id") + record.item_id = _codex_optional_string(record, item, "id") + if record.turn_id is None or record.item_id is None: + record.reject("codex_unmatched_conversation_event") + record.role = "human" if item_kind == "UserMessage" else "assistant" + record.phase = _codex_optional_string(record, item, "phase") + content = item.get("content") + expected_type = "text" if record.role == "human" else "Text" + if not isinstance(content, list) or any( + not isinstance(block, dict) or block.get("type") != expected_type + or not isinstance(block.get("text"), str) for block in content + ): + record.reject("codex_unmatched_conversation_event") + elif record.role == "assistant" and len(content) != 1: + record.reject("codex_unsupported_projection") + else: + record.projection = "".join(block["text"] for block in content) + if any(item.get(key) not in (None, []) for key in ("delivery", "questions", "memory_citation")): + record.reject("codex_unsupported_projection") + return record + if kind not in ("user_message", "agent_message"): + record.reject("codex_unrecognized_record") + return record + record.role = "human" if kind == "user_message" else "assistant" + record.phase = _codex_optional_string(record, payload, "phase") + if not isinstance(payload.get("message"), str): + record.reject("codex_unmatched_conversation_event") + else: + record.projection = payload["message"] + # These native event fields carry additional conversation unavailable to + # the plain-text decoder. Empty producer defaults do not add evidence. + media = ("images", "image_details", "file_ids", "file_id_details", "image_order", "local_images", + "local_image_details", "audio", "local_audio", "delivery", "questions", "memory_citation") + if any(payload.get(key) not in (None, []) for key in media): + record.reject("codex_unmatched_conversation_event") + return record + + +def _codex_echo_index(credits: list[tuple[int, _CodexRecord]], turn_id: str | None) -> dict[tuple, list[int]]: + """Eight bounded index entries per occurrence cover optional echo selectors.""" + index: dict[tuple, list[int]] = {} + for position, (_, candidate) in enumerate(credits): + for phase in (candidate.phase, _CODEX_ANY): + for turn in (turn_id or candidate.turn_id, _CODEX_ANY): + for item in (candidate.item_id, _CODEX_ANY): + key = (candidate.role, candidate.projection, phase, turn, item) + index.setdefault(key, []).append(position) + return index + + +def _codex_next_echo(index: dict[tuple, list[int]], event: _CodexRecord, cursor: int) -> int | None: + """Find the next compatible occurrence with at most two binary searches.""" + prefix = (event.role, event.projection, event.phase if event.phase is not None else _CODEX_ANY, + event.turn_id if event.turn_id is not None else _CODEX_ANY) + # Missing canonical IDs are compatible with a supplied event ID. An event + # without an ID queries all occurrences, not only those missing an ID. + item_keys = (_CODEX_ANY,) if event.item_id is None else (event.item_id, None) + matches: list[int] = [] + for item in item_keys: + positions = index.get((*prefix, item), ()) + offset = bisect_left(positions, cursor) + if offset < len(positions): + matches.append(positions[offset]) + return min(matches) if matches else None + + +def _reconcile_codex(records: list[tuple[int, _CodexRecord]]) -> None: + """Reconcile supplied-byte evidence; events never create canonical turns.""" + scopes: dict[int, list[tuple[int, _CodexRecord]]] = {} + scope_ids: dict[int, str | None] = {} + invalid_scopes: set[int] = set() + terminals: list[tuple[int, int, _CodexRecord]] = [] + scope = 0 + active = False + closed_ids: set[str] = set() + for index, record in records: + if record.kind == "session_meta": + scope += 1 + active = False + closed_ids.clear() + continue + if record.kind in _CODEX_START: + overlapping = active + if overlapping: + invalid_scopes.add(scope) + record.reject("codex_unsupported_lifecycle") + scope += 1 + active = True + scope_ids[scope] = record.turn_id + if overlapping or record.turn_id is None or record.malformed: + invalid_scopes.add(scope) + scopes.setdefault(scope, []) + continue + if record.kind in _CODEX_END | {"turn_aborted"}: + if active: + if record.turn_id != scope_ids.get(scope): + invalid_scopes.add(scope) + record.reject("codex_unsupported_lifecycle") + terminals.append((index, scope, record)) + active = False + elif record.turn_id in closed_ids or record.turn_id is None: + record.reject("codex_unsupported_lifecycle") + else: + # No start in this supplied segment: only explicit response + # turn metadata can explain a nonnull terminal summary. + terminals.append((index, scope, record)) + if record.turn_id is not None: + closed_ids.add(record.turn_id) + scope += 1 + continue + if active and record.turn_id is not None and record.turn_id != scope_ids.get(scope): + record.reject("codex_unsupported_lifecycle") + scopes.setdefault(scope, []).append((index, record)) + + for group, items in scopes.items(): + if group in invalid_scopes: + for _, record in items: + if record.role: + record.reject("codex_unsupported_lifecycle") + credits = [(index, record) for index, record in items if record.kind == "response_item" + and record.projection is not None and not record.malformed and not record.unrecognized] + credit_index = _codex_echo_index(credits, scope_ids.get(group)) + cursors: dict[str, int] = {} + for _, event in items: + if event.kind not in ("user_message", "agent_message", "item_completed") or not event.role: + continue + if event.malformed or event.unrecognized: + continue + position = _codex_next_echo(credit_index, event, cursors.get(event.role, 0)) + if position is None: + event.reject("codex_unmatched_conversation_event") + else: + cursors[event.role] = position + 1 + for index, group, terminal in terminals: + if terminal.projection is None or terminal.malformed or terminal.unrecognized: + continue + # Terminal evidence is independent of echo credit consumption. Only + # earlier canonical occurrences in this lifecycle may explain it. + candidates = scopes.get(group, []) + explicit_ids = {candidate.turn_id for before, candidate in candidates + if before < index and candidate.kind == "response_item" and candidate.role + and candidate.turn_id is not None} + ambiguous = group not in scope_ids and explicit_ids != {terminal.turn_id} + if ambiguous or group in invalid_scopes or not any( + before < index and candidate.kind == "response_item" and candidate.role == "assistant" + and candidate.projection == terminal.projection and not candidate.malformed and not candidate.unrecognized + and terminal.turn_id == (scope_ids.get(group) or candidate.turn_id) + for before, candidate in reversed(candidates) + ): + terminal.reject("codex_unmatched_terminal") + + def _transcript_path(source: str | Path) -> Path | None: if isinstance(source, Path): return source @@ -224,13 +644,20 @@ def _parse_transcript_text_outcome(text: str, *, jsonl: bool = False) -> Transcr messages: list[Message] = [] malformed = unrecognized = 0 - for entry in entries: + native_records: list[tuple[int, _CodexRecord]] = [] + for index, entry in enumerate(entries): if isinstance(entry, str) and allow_strings: messages.append(Message(role="unknown", content=entry)) continue if not isinstance(entry, dict): malformed += 1 continue + native = _decode_codex_record(entry) + if native is not None: + native_records.append((index, native)) + if native.message is not None: + messages.append(native.message) + continue top_type = entry.get("type") or entry.get("role") or "unknown" if not isinstance(top_type, str): malformed += 1 @@ -258,8 +685,12 @@ def _parse_transcript_text_outcome(text: str, *, jsonl: bool = False) -> Transcr continue if message is not None: messages.append(message) + _reconcile_codex(native_records) + malformed += sum(record.malformed for _, record in native_records) + unrecognized += sum(record.unrecognized for _, record in native_records) status = ("partial" if messages else "malformed") if malformed else "partial" if unrecognized else "complete" errors = (("malformed_records",) if malformed else ()) + (("unrecognized_content",) if unrecognized else ()) + errors += tuple(sorted({category for _, record in native_records for category in record.errors})) return TranscriptOutcome(messages, status, len(entries), malformed, errors, unrecognized_records=unrecognized) @@ -398,6 +829,10 @@ def _extract_message(entry: dict) -> Message | None: Non-conversation entries like ``file-history-snapshot`` and ``progress`` are filtered out. """ + native = _decode_codex_record(entry) + if native is not None: + return native.message + # Handle different field names across formats top_type = entry.get("type") or entry.get("role") or "unknown"