AIGTWY-4878 [1/3] Extract existing native transcript helpers

tests/README.md

@@ -139,6 +139,10 @@ Pi's token-command tests in `test_agent_pi.py` exercise Windows executable paths
 through POSIX parsing, including spaces, apostrophes, profile names, and PAT mode.
 They do not launch Pi or Git Bash on Windows.
 
+Shared native transcript parsing is checked by `test_integration_agent_helpers.py` and
+`test_integration_evidence.py`. These component checks protect existing journey evidence;
+they do not establish live agent coverage.
+
 ## CUJ coverage matrix
 
 These are **implemented assertions**, not a claim that every version passes.

tests/integration/README.md

@@ -99,6 +99,12 @@ when an explicit model is supplied. The only live OpenCode journey is the headle
 prompt case in `test_ug_opencode_headless.py`; there is no dedicated live OpenCode
 model-selection journey.
 
+## Shared agent helpers
+
+`utils/agents/claude.py` and `utils/agents/codex.py` own the existing native answer
+parsing, session paths, and child-session detection. `utils/evidence.py` dispatches
+through them without changing existing journey commands or completion assertions.
+
 ## Run a specific combination
 
 Prerequisites: Python 3.12+, uv, and Node/npm. Live runs also require Databricks

tests/integration/utils/agents/__init__.py

@@ -0,0 +1 @@
+"""Native agent transcript helpers used by the shared evidence layer."""

tests/integration/utils/agents/claude.py

@@ -0,0 +1,22 @@
+"""Claude transcript helpers for integration journeys."""
+
+SESSION_DIRECTORY = ".claude/projects"
+
+
+def is_child_session(path: str, records: list[dict]) -> bool:
+    del records
+    return "/subagents/" in path
+
+
+def assistant_answers(records: list[dict]) -> list[str]:
+    answers = []
+    for record in records:
+        if record.get("type") == "assistant":
+            message = record.get("message", {})
+            if message.get("role") == "assistant":
+                answers.extend(
+                    part["text"]
+                    for part in message.get("content", [])
+                    if part.get("type") == "text" and isinstance(part.get("text"), str)
+                )
+    return answers

tests/integration/utils/agents/codex.py

@@ -0,0 +1,23 @@
+"""Codex transcript helpers for integration journeys."""
+
+SESSION_DIRECTORY = ".codex/sessions"
+
+
+def is_child_session(path: str, records: list[dict]) -> bool:
+    del path
+    return any(
+        record.get("type") == "session_meta"
+        and isinstance(record.get("payload", {}).get("source"), dict)
+        and "subagent" in record["payload"]["source"]
+        for record in records
+    )
+
+
+def assistant_answers(records: list[dict]) -> list[str]:
+    answers = []
+    for record in records:
+        if record.get("type") == "event_msg":
+            payload = record.get("payload", {})
+            if payload.get("type") == "task_complete" and payload.get("last_agent_message"):
+                answers.append(payload["last_agent_message"])
+    return answers

tests/integration/utils/evidence.py

@@ -5,6 +5,10 @@
 import uuid
 from pathlib import Path
 
+from .agents import claude, codex
+
+_AGENT_HELPERS = {"claude": claude, "codex": codex}
+
 
 def assert_no_terminal_api_error(screen: str) -> None:
     """Fail on definitive client errors, not an in-progress transient retry."""
@@ -37,39 +41,19 @@ def read_jsonl(path: Path) -> list[dict]:
 
 
 def agent_sessions(session, agent: str) -> dict[str, list[dict]]:
-    directory = session.home / (".claude/projects" if agent == "claude" else ".codex/sessions")
+    directory = session.home / _AGENT_HELPERS.get(agent, codex).SESSION_DIRECTORY
     return {
         str(path.relative_to(directory)): read_jsonl(path) for path in directory.rglob("*.jsonl")
     }
 
 
 def assistant_answers(agent: str, records: list[dict]) -> list[str]:
-    answers = []
-    for record in records:
-        if agent == "claude" and record.get("type") == "assistant":
-            message = record.get("message", {})
-            if message.get("role") == "assistant":
-                answers.extend(
-                    part["text"]
-                    for part in message.get("content", [])
-                    if part.get("type") == "text" and isinstance(part.get("text"), str)
-                )
-        if agent == "codex" and record.get("type") == "event_msg":
-            payload = record.get("payload", {})
-            if payload.get("type") == "task_complete" and payload.get("last_agent_message"):
-                answers.append(payload["last_agent_message"])
-    return answers
+    helper = _AGENT_HELPERS.get(agent)
+    return helper.assistant_answers(records) if helper is not None else []
 
 
 def is_child_session(agent: str, path: str, records: list[dict]) -> bool:
-    if agent == "claude":
-        return "/subagents/" in path
-    return any(
-        record.get("type") == "session_meta"
-        and isinstance(record.get("payload", {}).get("source"), dict)
-        and "subagent" in record["payload"]["source"]
-        for record in records
-    )
+    return _AGENT_HELPERS.get(agent, codex).is_child_session(path, records)
 
 
 def assistant_answer_contains(session, agent: str, value: str, *, child: bool = False) -> bool:

tests/integration/utils/terminal.py

@@ -250,9 +250,22 @@ def boot(self, timeout=120):
                 ),
                 (
                     "trust-directory",
-                    self.session.cwd.name in text
-                    and "trust" in text.lower()
-                    and bool(re.search(r"1[.)]\s+Yes, (?:continue|proceed)", text)),
+                    self.agent == "codex"
+                    and bool(
+                        re.search(
+                            rf"(?m)^\s*>\s+You are in "
+                            rf"{re.escape(str(self.session.cwd.parent))}/[^/\r\n]*$",
+                            text,
+                        )
+                    )
+                    and (
+                        "Do you trust the contents of this directory?" in text
+                        or "directory allows project-local config, hooks, and exec policies to load."
+                        in text
+                    )
+                    and bool(re.search(r"(?m)^\s*[›❯>]\s*1[.)]\s+Yes, continue\s*$", text))
+                    and bool(re.search(r"(?m)^\s*2[.)]\s+No, quit\s*$", text))
+                    and "Press enter to continue" in text,
                     "\r",
                 ),
             ]
@@ -265,6 +278,7 @@ def boot(self, timeout=120):
                         handled.add(label)
                     break
             if matched:
+                ready_since = None
                 continue
             assert "Select login method:" not in text, (
                 "Configured ug launched Claude's account-login flow instead of its gateway session:\n"
@@ -278,6 +292,7 @@ def boot(self, timeout=120):
             if (
                 title in text
                 and "loading" not in text.lower()
+                and not re.search(r"(?m)^\s*>\s+You are in\b", text)
                 and re.search(r"(?m)^\s*[❯›>]\s*(?!\d+[.)])", text)
             ):
                 ready_since = ready_since or time.monotonic()