Integrate reviewed reliability and hot-path fixes

backend/app/chat.py

@@ -61,6 +61,7 @@
   StashToolOutput,
   alloc_run_token,
   await_ack as _await_ack,
+  cid_of,
   get_writer,
   next_message_ts as _next_message_ts,
   update_last_assistant_message as _update_last_assistant_message,
@@ -76,8 +77,10 @@
   excerpt_tool_output,
   finalize_blocks,
   process_event,
+  tool_output_exit_code,
   undo_question_scrub,
 )
+from app.memory_recall import recall_from_command, recall_from_result
 from app.providers import effective_agent_settings, get_provider, get_skill_path
 from app.runner_registry import registry
 from app.runtime_types import ChatEvent
@@ -226,6 +229,44 @@ def active_sink_memory_diagnostics(*, include_payloads: bool = True) -> list[dic
   return diagnostics
 
 
+def steered_into_turn_event(stored_messages: list[dict]) -> dict:
+  """Build the `steered_into_turn` SSE payload for a batch of steered rows.
+
+  `steered_into_turn` is the AUTHORITATIVE CUT and the client's ONLY "seal the
+  live stream here and re-base it" signal. It means the transcript split has
+  COMMITTED: A1 sealed, these rows appended after it, the sink reset for A2. So
+  it may only be published from the instant the split really happens — by the
+  Claude runner immediately after `split_for_steer`, and by the steer route for
+  Codex (whose `turn.steer()` has no interrupt boundary, so the route IS the
+  seal point). Two publishers, one builder, so the wire shape cannot drift.
+
+  Publishing it at HTTP arrival on the deferred (Claude) path is what made a
+  steer paint duplicated output for the rest of the turn: every block the runner
+  streamed between arrival and the real seal was accumulated into the sealed A1
+  AND left at the head of the client's freshly re-based stream. The deferred
+  path publishes NOTHING at arrival — the 202's own `pending_messages` is what
+  keeps the accepted row visible until the cut, so there is no second channel
+  reconciling the same tray.
+  """
+  return {
+    "type": "steered_into_turn",
+    "messages": [
+      {
+        "role": "user",
+        "ts": msg.get("ts"),
+        "cid": cid_of(msg),
+        "content": msg.get("content", ""),
+        **({"attachments": msg.get("attachments")} if msg.get("attachments") else {}),
+      }
+      for msg in stored_messages
+    ],
+    # Backward-compatible shape for any existing client still expecting a
+    # single steered row.
+    "ts": stored_messages[-1].get("ts"),
+    "content": stored_messages[-1].get("content", ""),
+  }
+
+
 class _ChatEventSink:
   """Bridges SDK-runner events to broadcast + the chat-writer actor.
 
@@ -389,7 +430,71 @@ def _log_if_failed(fut, _kind=type(cmd).__name__, _cid=self.chat_id):
 
     ack.add_done_callback(_log_if_failed)
 
-  def _reduce_tool_output(self, event: ChatEvent) -> None:
+  def _memory_recall_for_tool(self, tool_use_id) -> dict | None:
+    """Return the input-time Memory marker for this tool, if there is one.
+
+    Read-only by design: resolving the block is a question, not the place to
+    adopt a legacy id (`process_event` still owns that a moment later). Only a
+    tool whose COMMAND named memory_search may go on to cite notes, so output
+    text alone can never mint a citation.
+    """
+    for blk in reversed(self.assistant_blocks):
+      if blk.get("type") != "tool":
+        continue
+      if tool_use_id:
+        if blk.get("tool_use_id") == tool_use_id:
+          recall = blk.get("recall")
+          return recall if isinstance(recall, dict) else None
+        continue
+      # Legacy events without an id: the newest still-open tool is the only
+      # safe candidate, matching `_tool_block_for_event`'s fallback.
+      if blk.get("status") != "done":
+        recall = blk.get("recall")
+        return recall if isinstance(recall, dict) else None
+    return None
+
+  def _tool_was_memory_recall(self, tool_use_id) -> bool:
+    return self._memory_recall_for_tool(tool_use_id) is not None
+
+  def _stamp_memory_recall(self, event: ChatEvent) -> None:
+    """Name a Memory-app recall on the event, in two lifecycle phases.
+
+    The documented simple command identifies the lookup, so the live turn can
+    say it is remembering while the search runs. Only the final output event
+    settles it from the Memory app's structured result; streaming deltas cannot
+    prematurely claim success, emptiness, or failure.
+    """
+    if event.get("type") in ("tool_start", "tool_input"):
+      if event.get("type") == "tool_start" and event.get("tool") != "Bash":
+        return
+      # Both a tool_start AND a tool_input can arrive for one tool call on the
+      # Claude runner. Stamp the command-derived marker exactly once per block:
+      # if the block for this tool_use_id already carries a recall marker, leave
+      # it settled and skip. (Codex has no tool_input; Claude's tool_start input
+      # is empty, so in practice only one phase produces a marker — this keeps a
+      # future runner that populates both from double-stamping.)
+      if self._tool_was_memory_recall(event.get("tool_use_id")):
+        return
+      recall = recall_from_command(event.get("input"))
+      if recall is not None:
+        event["recall"] = recall
+      return
+    pending = self._memory_recall_for_tool(event.get("tool_use_id"))
+    if event.get("output_complete") and pending is not None:
+      settled = recall_from_result(
+        event.get("content"), event.get("output_exit_code"),
+      )
+      # The command path is the authoritative installed-app identity. Stamp it
+      # onto each successful note so deep links keep working when the official
+      # system app had to install as memory-2 (or another numeric suffix).
+      app_slug = pending.get("app_slug")
+      if settled.get("status") == "hit" and isinstance(app_slug, str):
+        settled["notes"] = [
+          {**note, "app_slug": app_slug} for note in settled.get("notes", [])
+        ]
+      event["recall"] = settled
+
+  def _reduce_tool_output(self, event: ChatEvent) -> bool:
     """Move a large tool_output's full text OFF the wire (contract rule 6).
 
     This is the single funnel where the live SSE push, the catch-up event_log,
@@ -416,11 +521,11 @@ def _reduce_tool_output(self, event: ChatEvent) -> None:
     content = event.get("content")
     if (not isinstance(content, str)
         or len(content) <= TOOL_OUTPUT_INLINE_THRESHOLD):
-      return
+      return False
     if not self.chat_id:
       # No chat to key a stash by (a detached/synthetic sink — chat_id is always
       # set on the live path). Can't move the text off-wire safely, so leave it.
-      return
+      return False
     tool_use_id = event.get("tool_use_id")
     if not tool_use_id:
       # Unexpected post-card-221: mint a stash id and stamp it on the event so
@@ -432,16 +537,21 @@ def _reduce_tool_output(self, event: ChatEvent) -> None:
         "minted stash id %s", self.chat_id, tool_use_id,
       )
     full = content
-    excerpt, full_len, exit_code = excerpt_tool_output(full)
+    excerpt, full_len, parsed_exit_code = excerpt_tool_output(full)
     event["content"] = excerpt
     event["output_truncated"] = True
     event["output_full_len"] = full_len
-    event["output_exit_code"] = exit_code
+    # Codex can supply a typed exit code independently of its display text.
+    # That runner-owned fact outranks best-effort parsing of the excerpt.
+    typed_exit_code = event.get("output_exit_code")
+    if not isinstance(typed_exit_code, int) or isinstance(typed_exit_code, bool):
+      event["output_exit_code"] = parsed_exit_code
     self._submit_fire_and_forget(
       StashToolOutput(
         chat_id=self.chat_id, tool_use_id=tool_use_id, output=full,
       )
     )
+    return True
 
   def record_lifecycle(self, event: dict) -> None:
     """Queue private lifecycle metadata without broadcasting it.
@@ -505,8 +615,19 @@ def publish(self, event: ChatEvent) -> bool:
     # its full text BEFORE process_event (which copies content onto the block)
     # and before the broadcast below, so the rewritten event is the single
     # source feeding the persisted block, the live wire, and the catch-up log.
+    #
+    # Reduce first so a large JSON envelope is parsed only once. The app prints
+    # its bounded structured Memory result last, so the carved tail still
+    # contains the line that settles a recognized lookup.
+    output_reduced = False
     if event_type == "tool_output":
-      self._reduce_tool_output(event)
+      output_reduced = self._reduce_tool_output(event)
+      if not output_reduced and event.get("output_exit_code") is None:
+        exit_code = tool_output_exit_code(event.get("content"))
+        if exit_code is not None:
+          event["output_exit_code"] = exit_code
+    if event_type in ("tool_start", "tool_input", "tool_output"):
+      self._stamp_memory_recall(event)
     if event_type == "thinking":
       self._prepare_thinking_event(event)
 

backend/app/chat_transcript.py

@@ -4,6 +4,14 @@
 
 import re
 
+from app.memory_recall import (
+  RECALL_EMPTY,
+  RECALL_FAILED,
+  RECALL_HIT,
+  RECALL_SEARCHING,
+  merge_recall_notes,
+)
+
 
 _QUESTION_TOOLS = {"AskUserQuestion", "request_user_input"}
 _IMAGE_PATH_RE = re.compile(
@@ -116,7 +124,14 @@ def historical_tool_output_ids(
 
 def _distinctive_activity(block: dict) -> bool:
   """Keep notable one-line activity beats out of a folded metadata run."""
-  if block.get("type") != "tool" or block.get("tool") != "Read":
+  if block.get("type") != "tool":
+    return False
+  # Consulting Memory is a beat worth seeing on its own, not shell housekeeping
+  # folded into "ran commands". The marker was stamped from the command itself,
+  # so this needs no knowledge of how that command is spelled.
+  if isinstance(block.get("recall"), dict):
+    return True
+  if block.get("tool") != "Read":
     return False
   raw = block.get("input")
   if isinstance(raw, dict):
@@ -161,6 +176,11 @@ def _compact_activity_item(block: dict) -> dict:
   # tool input remains in the on-demand activity detail.
   if block.get("tool") == "Read" and isinstance(block.get("input"), str):
     tool["input"] = block["input"][:2048]
+  # A Memory recall is already a bounded citation set, and it is what the
+  # collapsed line says ("Recalled 4 notes from Memory"). Dropping it here
+  # would make the beat visible live and gone on the next chat load.
+  if isinstance(block.get("recall"), dict):
+    tool["recall"] = block["recall"]
   return tool
 
 
@@ -242,6 +262,31 @@ def _compact_activity_run(
     if len(sources) >= _MAX_COMPACT_SOURCES:
       break
 
+  # Memory citations roll up for the same reason web sources do: the message
+  # renders them once per turn, so they must outlive the individual tool blocks
+  # this projection folds away. `_compact_activity_entries` keeps only two
+  # entries per tool name, so without this a third lookup's notes would vanish.
+  recall_notes: list[dict] = []
+  seen_recall_paths: set[str] = set()
+  recall_status = ""
+  recall_rank = {
+    RECALL_SEARCHING: 0,
+    RECALL_FAILED: 1,
+    RECALL_EMPTY: 2,
+    RECALL_HIT: 3,
+  }
+  for _, block in blocks:
+    recall = block.get("recall")
+    if not isinstance(recall, dict):
+      continue
+    # A real hit outranks an empty search, which outranks a failed probe. This
+    # preserves useful evidence without letting one failure erase a successful
+    # result elsewhere in the same folded run.
+    status = recall.get("status")
+    if recall_rank.get(status, -1) > recall_rank.get(recall_status, -1):
+      recall_status = status
+    merge_recall_notes(recall_notes, seen_recall_paths, recall)
+
   start = blocks[0][0]
   end = blocks[-1][0] + 1
   return {
@@ -255,6 +300,10 @@ def _compact_activity_run(
       block.get("type") == "tool" for _, block in blocks
     ),
     **({"sources": sources} if sources else {}),
+    **(
+      {"recall": {"status": recall_status, "notes": recall_notes}}
+      if recall_status else {}
+    ),
   }