Replay observations and auto-approve pending calls on agent resume

- Add RunStore.observations_for_run to extract completed tool call
  results from a persisted run for replay into a new run's context
- Add RunStore.pending_approval_for_run to detect last unresolved
  approval request, auto-approving it on resume
- Pre-populate context observations from the prior run and count
  replayed calls against the tool call budget
- Add --fresh-restart flag to skip replay and re-run from scratch

Author: johnteee

AGENTS.md

@@ -28,13 +28,13 @@
 <claude-mem-context>
 # Memory Context
 
-# [teaagent] recent context, 2026-05-08 8:57am GMT+8
+# [teaagent] recent context, 2026-05-08 2:07pm GMT+8
 
 Legend: 🎯session 🔴bugfix 🟣feature 🔄refactor ✅change 🔵discovery ⚖️decision 🚨security_alert 🔐security_note
 Format: ID TIME TYPE TITLE
 Fetch details: get_observations([IDs]) | Search: mem-search skill
 
-Stats: 18 obs (7,486t read) | 161,442t work | 95% savings
+Stats: 19 obs (8,027t read) | 170,927t work | 95% savings
 
 ### May 8, 2026
 22 12:31a 🟣 TeaAgent P0 Agent Harness — Initial Implementation
@@ -61,24 +61,7 @@ S7 Generate commit message for staged changes — TeaAgent intent clarification
 38 8:39a 🟣 Deterministic task clarification system added to TeaAgent
 39 8:46a 🟣 Memory Catalog added to teaagent
 S8 Add workspace memory catalog to teaagent — new MemoryCatalog feature with CLI, TUI, and agent prompt injection (May 8 at 8:46 AM)
-**Investigated**: Staged git diff across all modified files in /Users/teee/dev/teaagent, covering the full scope of the memory catalog implementation.
+40 8:57a 🟣 Deterministic task-based model routing added to teaagent
 
-**Learned**: teaagent stores workspace memories as append-only JSONL at .teaagent/memory.jsonl. Search uses case-insensitive token intersection (all query tokens must appear in content+tags haystack). Agent runs automatically inject up to 5 matching memories into the model context dict under a "memories" key, which assemble_agent_prompt passes through alongside "observations" and "task_spec".
-
-**Completed**: - teaagent/memory.py created: MemoryEntry (frozen dataclass, uuid4 hex id, content, tags, created_at) and MemoryCatalog (add/list/search/show, JSONL persistence), plus normalize_tags, memory_matches, memory_entries_to_prompt helpers
-    - teaagent/chat_agent.py: ChatAgentConfig.memory_limit=5 added; run_chat_agent searches catalog at task start and wraps engine.decide with with_memories() injector
-    - teaagent/cli.py: memory subcommand with add/list/search/show sub-subcommands, --root and --tag/--limit flags
-    - teaagent/tui.py: memory command routed to _handle_memory(); HELP_TEXT updated
-    - teaagent/prompt.py: memories field threaded through assemble_agent_prompt context
-    - teaagent/__init__.py: MemoryCatalog and MemoryEntry exported in __all__
-    - tests/test_memory.py: new file with catalog unit test and CLI integration test
-    - tests/test_chat_agent.py: memory injection end-to-end test added
-    - tests/test_tui.py: TUI memory command test added
-    - docs/cli.md: Memory Catalog section with bash and TUI usage examples
-    - Commit message generated for all staged changes
-
-**Next Steps**: Session appears complete — commit message was the final deliverable. No further work indicated in the observed session.
-
-
-Access 161k tokens of past work via get_observations([IDs]) or mem-search skill.
+Access 171k tokens of past work via get_observations([IDs]) or mem-search skill.
 </claude-mem-context>

docs/cli.md

@@ -452,6 +452,10 @@ Resume the original task from a persisted run id with optional new approval toke
 teaagent agent resume gpt <run_id> --root /path/to/repo --approve-call-id write-1
 ```
 
+By default, resume replays already-completed `tool_call_completed` observations into the new run's context so the model does not have to redo prior tool calls. If the original run paused with `pending_approval`, the pending `call_id` is auto-added to the approval list and reported back as `auto_approved_call_id` in the response payload.
+
+Pass `--fresh-restart` to skip replay and re-run the original task from scratch.
+
 Inside TUI:
 
 ```text

teaagent/chat_agent.py

@@ -97,6 +97,7 @@ def run_chat_agent(
     registry: Optional[ToolRegistry] = None,
     task_spec: Optional[str] = None,
     depth: int = 0,
+    initial_observations: Optional[list[dict[str, Any]]] = None,
 ) -> RunResult:
     tool_registry = registry or build_workspace_tool_registry(config.root)
     if config.enable_subagent and depth < config.max_subagent_depth:
@@ -135,6 +136,7 @@ def run_chat_agent(
             task=task,
             decide=lambda context: engine.decide(with_memories(context, memories)),
             run_id=run_id,
+            initial_observations=initial_observations,
         )
     finally:
         if heartbeat is not None:

teaagent/cli.py

@@ -185,6 +185,11 @@ def build_parser() -> argparse.ArgumentParser:
     agent_resume.add_argument("--subagent", action="store_true", help="Expose the 'subagent' tool.")
     agent_resume.add_argument("--max-subagent-depth", type=int, default=1, help="Maximum nested subagent depth.")
     agent_resume.add_argument("--heartbeat", type=float, default=0.0, help="Heartbeat interval seconds. 0 disables.")
+    agent_resume.add_argument(
+        "--fresh-restart",
+        action="store_true",
+        help="Re-run the original task from scratch instead of replaying observations from the prior run.",
+    )
     agent_resume.set_defaults(func=agent_resume_command)
 
     agent_status = agent_subparsers.add_parser("status", help="Show liveness status of a persisted run.")
@@ -387,10 +392,33 @@ def agent_resume_command(args: argparse.Namespace) -> int:
     except (FileNotFoundError, ValueError) as exc:
         print_json({"status": "error", "message": str(exc)})
         return 1
-    return _execute_agent_task(args, original_task, resumed_from=args.run_id)
-
 
-def _execute_agent_task(args: argparse.Namespace, task: str, *, resumed_from: Optional[str] = None) -> int:
+    initial_observations: list[dict[str, Any]] = []
+    auto_approved: Optional[str] = None
+    if not args.fresh_restart:
+        initial_observations = store.observations_for_run(args.run_id)
+        pending = store.pending_approval_for_run(args.run_id)
+        if pending and pending["call_id"] not in args.approve_call_id:
+            args.approve_call_id = list(args.approve_call_id) + [pending["call_id"]]
+            auto_approved = pending["call_id"]
+
+    return _execute_agent_task(
+        args,
+        original_task,
+        resumed_from=args.run_id,
+        initial_observations=initial_observations,
+        auto_approved_call_id=auto_approved,
+    )
+
+
+def _execute_agent_task(
+    args: argparse.Namespace,
+    task: str,
+    *,
+    resumed_from: Optional[str] = None,
+    initial_observations: Optional[list[dict[str, Any]]] = None,
+    auto_approved_call_id: Optional[str] = None,
+) -> int:
     task_spec = None
     if args.clarify:
         clarification = clarify_task(task)
@@ -423,12 +451,17 @@ def _execute_agent_task(args: argparse.Namespace, task: str, *, resumed_from: Op
         ),
         audit=audit,
         task_spec=task_spec,
+        initial_observations=initial_observations,
     )
     store.logger_for_result(result, audit)
     payload = run_result_payload(result, routing=routing.to_dict() if routing else None)
     if resumed_from:
         payload["resumed_from"] = resumed_from
         payload["task"] = task
+        if initial_observations:
+            payload["replayed_observations"] = len(initial_observations)
+        if auto_approved_call_id is not None:
+            payload["auto_approved_call_id"] = auto_approved_call_id
     print_json(payload)
     return 0 if result.status == "completed" else 1
 

teaagent/run_store.py

@@ -73,6 +73,39 @@ def task_for_run(self, run_id: str) -> str:
                     return task
         raise ValueError(f"run '{run_id}' has no run_started task")
 
+    def observations_for_run(self, run_id: str) -> list[dict[str, Any]]:
+        observations: list[dict[str, Any]] = []
+        for event in self.show_run(run_id):
+            if event.get("event_type") != "tool_call_completed":
+                continue
+            payload = event.get("payload") or {}
+            call_id = payload.get("call_id")
+            tool_name = payload.get("tool_name")
+            result = payload.get("result")
+            if isinstance(call_id, str) and isinstance(tool_name, str) and isinstance(result, dict):
+                observations.append({"call_id": call_id, "tool_name": tool_name, "result": result})
+        return observations
+
+    def pending_approval_for_run(self, run_id: str) -> Optional[dict[str, Any]]:
+        pending: Optional[dict[str, Any]] = None
+        for event in self.show_run(run_id):
+            event_type = event.get("event_type")
+            payload = event.get("payload") or {}
+            if event_type == "tool_call_pending_approval":
+                call_id = payload.get("call_id")
+                tool_name = payload.get("tool_name")
+                arguments = payload.get("arguments")
+                if isinstance(call_id, str) and isinstance(tool_name, str):
+                    pending = {
+                        "call_id": call_id,
+                        "tool_name": tool_name,
+                        "arguments": arguments if isinstance(arguments, dict) else {},
+                    }
+            elif event_type in {"tool_call_approved", "tool_call_denied", "run_completed", "run_failed"}:
+                if pending and payload.get("call_id") == pending.get("call_id"):
+                    pending = None
+        return pending
+
     def heartbeat_for_run(self, run_id: str) -> dict[str, Any]:
         events = self.show_run(run_id)
         last_heartbeat: Optional[dict[str, Any]] = None

teaagent/runner.py

@@ -91,13 +91,26 @@ def _assert_cost_budget(self, cost_cents: float) -> None:
         if cost_cents > self.budget.max_estimated_cost_cents:
             raise BudgetExceededError("cost budget exceeded")
 
-    def run(self, *, task: str, decide: DecisionFn, run_id: Optional[str] = None) -> RunResult:
+    def run(
+        self,
+        *,
+        task: str,
+        decide: DecisionFn,
+        run_id: Optional[str] = None,
+        initial_observations: Optional[list[dict[str, Any]]] = None,
+    ) -> RunResult:
         current_run_id = run_id or uuid4().hex
-        context: dict[str, Any] = {"task": task, "observations": []}
+        observations: list[dict[str, Any]] = list(initial_observations) if initial_observations else []
+        context: dict[str, Any] = {"task": task, "observations": observations}
         iterations = 0
-        tool_calls = 0
+        tool_calls = len(observations)
         cost_cents = 0.0
-        self.audit.record("run_started", current_run_id, task=task)
+        self.audit.record(
+            "run_started",
+            current_run_id,
+            task=task,
+            replayed_observations=len(observations),
+        )
 
         while iterations < self.budget.max_iterations:
             iterations += 1

tests/test_chat_agent.py

@@ -289,6 +289,74 @@ def test_cli_agent_resume_replays_original_task(self) -> None:
             self.assertEqual(payload["task"], "summarize repo")
             self.assertEqual(payload["final_answer"], "second")
 
+    def test_cli_agent_resume_replays_observations_and_auto_approves_pending(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            (Path(tmp) / "note.txt").write_text("hello", encoding="utf-8")
+            first_adapter = FakeAdapter(
+                [
+                    '{"type":"tool","tool_name":"workspace_read_file","arguments":{"path":"note.txt"},"call_id":"read-1"}',
+                    '{"type":"tool","tool_name":"workspace_write_file","arguments":{"path":"out.txt","content":"hello!"},"call_id":"write-1"}',
+                ]
+            )
+            with (
+                patch("teaagent.cli.create_llm_adapter", return_value=first_adapter),
+                redirect_stdout(io.StringIO()) as first_out,
+            ):
+                first_code = main(["agent", "run", "gpt", "process notes", "--root", tmp])
+            first_payload = json.loads(first_out.getvalue())
+            self.assertEqual(first_code, 1)
+            self.assertEqual(first_payload["status"], "pending_approval")
+            self.assertEqual(first_payload["approval"]["call_id"], "write-1")
+            run_id = first_payload["run_id"]
+
+            resume_adapter = FakeAdapter(
+                [
+                    '{"type":"tool","tool_name":"workspace_write_file","arguments":{"path":"out.txt","content":"hello!"},"call_id":"write-1"}',
+                    '{"type":"final","content":"wrote"}',
+                ]
+            )
+            with (
+                patch("teaagent.cli.create_llm_adapter", return_value=resume_adapter),
+                redirect_stdout(io.StringIO()) as resume_out,
+            ):
+                resume_code = main(["agent", "resume", "gpt", run_id, "--root", tmp])
+            resume_payload = json.loads(resume_out.getvalue())
+
+            self.assertEqual(resume_code, 0)
+            self.assertEqual(resume_payload["status"], "completed")
+            self.assertEqual(resume_payload["resumed_from"], run_id)
+            self.assertEqual(resume_payload["replayed_observations"], 1)
+            self.assertEqual(resume_payload["auto_approved_call_id"], "write-1")
+            self.assertEqual((Path(tmp) / "out.txt").read_text(encoding="utf-8"), "hello!")
+
+    def test_cli_agent_resume_fresh_restart_skips_replay(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            (Path(tmp) / "note.txt").write_text("hi", encoding="utf-8")
+            first_adapter = FakeAdapter(
+                [
+                    '{"type":"tool","tool_name":"workspace_read_file","arguments":{"path":"note.txt"},"call_id":"read-1"}',
+                    '{"type":"final","content":"first"}',
+                ]
+            )
+            with (
+                patch("teaagent.cli.create_llm_adapter", return_value=first_adapter),
+                redirect_stdout(io.StringIO()) as first_out,
+            ):
+                main(["agent", "run", "gpt", "read note", "--root", tmp])
+            run_id = json.loads(first_out.getvalue())["run_id"]
+
+            resume_adapter = FakeAdapter(['{"type":"final","content":"fresh"}'])
+            with (
+                patch("teaagent.cli.create_llm_adapter", return_value=resume_adapter),
+                redirect_stdout(io.StringIO()) as resume_out,
+            ):
+                main(["agent", "resume", "gpt", run_id, "--root", tmp, "--fresh-restart"])
+            payload = json.loads(resume_out.getvalue())
+
+            self.assertEqual(payload["final_answer"], "fresh")
+            self.assertNotIn("replayed_observations", payload)
+            self.assertNotIn("auto_approved_call_id", payload)
+
     def test_cli_agent_resume_unknown_run_id_returns_error(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             output = io.StringIO()

tests/test_p0_harness.py

@@ -190,6 +190,57 @@ def decide(context):
         self.assertNotIn("tool_call_started", [event.event_type for event in audit.events])
         self.assertEqual(audit.events[-1].payload["cost_cents"], 2.0)
 
+    def test_initial_observations_replayed_into_context(self) -> None:
+        audit = AuditLogger()
+        runner = AgentRunner(registry=build_registry(), audit=audit)
+        seen: list[list[dict]] = []
+
+        def decide(context):
+            seen.append(list(context["observations"]))
+            return FinalAnswer(content="ok")
+
+        replayed = [
+            {"call_id": "r1", "tool_name": "pilot_echo", "result": {"value": "earlier"}},
+            {"call_id": "r2", "tool_name": "pilot_echo", "result": {"value": "later"}},
+        ]
+        result = runner.run(
+            task="resume",
+            decide=decide,
+            run_id="run-replay",
+            initial_observations=replayed,
+        )
+
+        self.assertEqual(result.status, "completed")
+        self.assertEqual(result.tool_calls, 2)
+        self.assertEqual(seen[0], replayed)
+        run_started = next(e for e in audit.events if e.event_type == "run_started")
+        self.assertEqual(run_started.payload["replayed_observations"], 2)
+
+    def test_initial_observations_count_against_tool_call_budget(self) -> None:
+        audit = AuditLogger()
+        runner = AgentRunner(
+            registry=build_registry(),
+            audit=audit,
+            budget=RunBudget(max_iterations=3, max_tool_calls=2),
+        )
+
+        def decide(_context):
+            return ToolRequest(tool_name="pilot_echo", arguments={"value": "next"})
+
+        replayed = [
+            {"call_id": "r1", "tool_name": "pilot_echo", "result": {"value": "a"}},
+            {"call_id": "r2", "tool_name": "pilot_echo", "result": {"value": "b"}},
+        ]
+        result = runner.run(
+            task="budget-replay",
+            decide=decide,
+            run_id="run-budget-replay",
+            initial_observations=replayed,
+        )
+
+        self.assertEqual(result.status, "failed:model_logic")
+        self.assertEqual(result.tool_calls, 2)
+
     def test_cost_budget_blocks_final_after_decision_cost_update(self) -> None:
         audit = AuditLogger()
         runner = AgentRunner(