diff --git a/backend/app/chat.py b/backend/app/chat.py index a96465f7a..8bcd99252 100644 --- a/backend/app/chat.py +++ b/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) diff --git a/backend/app/chat_transcript.py b/backend/app/chat_transcript.py index 4da520fbd..a44432131 100644 --- a/backend/app/chat_transcript.py +++ b/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 {} + ), } diff --git a/backend/app/claude_sdk_runner.py b/backend/app/claude_sdk_runner.py index 1e61436b5..1fb60ee2d 100644 --- a/backend/app/claude_sdk_runner.py +++ b/backend/app/claude_sdk_runner.py @@ -408,13 +408,26 @@ async def interrupt(self) -> None: bound at the call site; this inner timeout protects any other direct caller. - Stop is the hard, immediate-cut path: it drops any buffered steer - (clearing `pending_steer` + `_steer_requested` so no boundary cut or - requery fires for work the user just abandoned) and interrupts the - live turn right now, without waiting for a content-block boundary. + Stop is the hard, immediate-cut path: it drops the buffered steer + ENTIRELY — the provider-facing text (`pending_steer` + `_steer_requested`, + so no boundary cut or requery fires for work the user just abandoned) AND + the transcript-side rows (`_steer_user_msgs` + `_steer_consume_cids`, so the + turn-end seal appends nothing). + + Both halves have to go, because Stop OWNS those rows from here on: a + deferred steer's row is still a durable entry in `chat.pending_messages` + (the split that would consume it never ran), `/chat/stop` clears that queue + and reports the cleared cids, and the client re-sends exactly them as one + fresh turn. Leaving the rows buffered meant the dying turn's seal appended + the same row into the transcript while the client re-sent it — the row + appeared twice, once interrupted and once answered. Nothing is lost by + dropping them here: they were never in the transcript, and Stop's own + clear-and-resend path is what preserves them. """ self.pending_steer = [] self._steer_requested = False + self._steer_user_msgs = [] + self._steer_consume_cids = [] await self._client.interrupt() try: await asyncio.wait_for(asyncio.shield(self._finished), timeout=5.0) @@ -506,12 +519,16 @@ async def _seal_steer_split(bc, active_client, chat_id: str) -> None: Called at each requery boundary (so A1 is sealed before the answer A2 streams) AND unconditionally in the turn-end `finally` (so a steer that was - buffered but never sealed — an exception/early-return before the requery, or - a hard Stop that cleared `pending_steer` — is still persisted rather than - discarded with the handle). A1 is the sink's accumulated pre-interrupt - content — complete once the turn closes — so `split_for_steer` seals it as - its own message, appends the steered row(s) after it, and resets the sink so - A2 lands fresh: reload order Q1, A1, Q2, A2. + buffered but never sealed — an exception/early-return before the requery — is + still persisted rather than discarded with the handle). A hard Stop is NOT one + of those cases: `interrupt()` drops the buffered rows outright because Stop's + clear-and-resend path owns them from that point (see `interrupt`), so the + finally finds an empty buffer and appends nothing to the turn it just killed. + + A1 is the sink's accumulated pre-interrupt content — complete once the turn + closes — so `split_for_steer` seals it as its own message, appends the steered + row(s) after it, and resets the sink so A2 lands fresh: reload order Q1, A1, + Q2, A2. This is the fix for the steer-merge: the route cannot know where A1 ends (at HTTP arrival A1 has not streamed yet, so a route-side split sealed an empty @@ -519,6 +536,16 @@ async def _seal_steer_split(bc, active_client, chat_id: str) -> None: does. `bc` is the live `_ChatEventSink`; a non-sink `bc` (legacy path / a test double) cannot persist here and drops the buffered rows. + This is ALSO where the client's cut lands. `steered_into_turn` is the client's + only "seal the live stream here and re-base it" signal, so it must be + published from the same instant as the durable split — deferring the split to + here while the route published the cut at HTTP arrival meant every block + streamed in between was BOTH folded into the sealed A1 and left at the head of + the client's re-based stream, painting twice for the rest of the turn. No + split (no live sink, or a failed write) publishes no cut: the client must + never re-base earlier than the server's actual seal. A split with no + publisher is the one asymmetric case — it commits and logs, see below. + Durability contract (adversarial-review hardening): - The rows are snapshotted BEFORE the await and only the snapshotted count is removed on success, so a second steer landing during `split_for_steer`'s @@ -535,6 +562,27 @@ async def _seal_steer_split(bc, active_client, chat_id: str) -> None: return consume = list(active_client._steer_consume_cids) split = getattr(bc, "split_for_steer", None) + # Resolve the client-facing publisher BEFORE committing anything. Take the + # broadcast off the SINK rather than re-resolving it by chat_id: the cut + # belongs in the same event log that carries A1's blocks (so a reconnect + # replays the boundary at its true position), and a lookup could hand back a + # successor turn's broadcast when this runs from the turn-end `finally`. + # + # A missing publisher does NOT abort the split: the rows are already durable + # in the pending queue and the client was told the steer landed, so + # persistence wins over notification. It does mean this seal produces no cut, + # leaving the client's live stream un-rebased until its next authoritative + # fetch — a real divergence, so it is logged loudly here rather than returned + # away silently after the write has already committed. + raw_bc = getattr(bc, "bc", None) + if raw_bc is not None and not callable(getattr(raw_bc, "publish", None)): + raw_bc = None + if split is not None and raw_bc is None: + log.error( + "steer split has no broadcast to publish the cut on chat_id=%s; " + "the transcript will be split but the client stream cannot re-base " + "until it refetches", chat_id, + ) if split is None: # No live sink (legacy/test caller): there is no streamed A1 to seal # against and no way to persist here — drop the buffer. @@ -544,7 +592,7 @@ async def _seal_steer_split(bc, active_client, chat_id: str) -> None: ) return try: - await split(rows, consume) + stored_result = await split(rows, consume) except Exception: # Leave the buffer intact so the turn-end finally retries the write. log.exception( @@ -557,6 +605,33 @@ async def _seal_steer_split(bc, active_client, chat_id: str) -> None: active_client._steer_consume_cids = ( active_client._steer_consume_cids[len(consume):] ) + # Publish the cut now that A1 + Q2 are committed, on the broadcast resolved + # above. No await separates the split from this publish, so no continuation + # block can slip in front of it. + if raw_bc is None: + return + from app.chat import steered_into_turn_event + + stored_messages = ( + stored_result.get("stored_messages") if isinstance(stored_result, dict) + else None + ) + if not isinstance(stored_messages, list) or not stored_messages: + # The writer echoes the rows it stored; fall back to the rows we handed it + # so an older/leaner ack shape still produces a well-formed cut. + stored_messages = rows + try: + raw_bc.publish(steered_into_turn_event(stored_messages)) + except Exception: + # The split already COMMITTED, so failing to announce it is a notification + # loss, not a durability one — same asymmetry as the missing-publisher case + # above. Swallow and log: this function is awaited from the turn-end + # `finally`, where a raise would skip unregistering the handle and + # disconnecting the client, leaving the chat looking permanently live. + log.exception( + "publishing the steer cut failed chat_id=%s; the split committed but the " + "client stream cannot re-base until it refetches", chat_id, + ) def _skill_file_read_name( @@ -1022,6 +1097,7 @@ def dispatch_sdk_message( "type": "tool_output", "content": output, "tool_use_id": block.tool_use_id, + "output_complete": True, }) if output.startswith("Web search results for query"): sources = sources_from_websearch_text(output) diff --git a/backend/app/codex_sdk_runner.py b/backend/app/codex_sdk_runner.py index 9bd90b298..9e20e0c1e 100644 --- a/backend/app/codex_sdk_runner.py +++ b/backend/app/codex_sdk_runner.py @@ -569,7 +569,10 @@ def _sdk_imports() -> dict[str, Any]: """ from openai_codex import ApprovalMode, AsyncCodex, Sandbox from openai_codex.client import CodexConfig - from openai_codex.errors import CodexRpcError, InvalidParamsError + from openai_codex.errors import ( + CodexRpcError, + TransportClosedError, + ) from openai_codex.types import ReasoningEffort, ReasoningSummary from openai_codex.generated.v2_all import ( AgentMessageDeltaNotification, @@ -646,7 +649,6 @@ def _sdk_imports() -> dict[str, Any]: "ErrorNotification": ErrorNotification, "FileChangePatchUpdatedNotification": FileChangePatchUpdatedNotification, "FileChangeThreadItem": FileChangeThreadItem, - "InvalidParamsError": InvalidParamsError, "ReasoningEffort": ReasoningEffort, "ReasoningSummary": ReasoningSummary, "Sandbox": Sandbox, @@ -667,6 +669,7 @@ def _sdk_imports() -> dict[str, Any]: "ThreadTokenUsageUpdatedNotification": ( ThreadTokenUsageUpdatedNotification ), + "TransportClosedError": TransportClosedError, "TurnCompletedNotification": TurnCompletedNotification, "TurnStatus": TurnStatus, "WebSearchThreadItem": WebSearchThreadItem, @@ -1193,9 +1196,13 @@ def _tool_completed_events(item: Any, sdk: dict[str, Any]) -> list[dict[str, Any if isinstance(item, sdk["CommandExecutionThreadItem"]): output = (item.aggregated_output or "").strip() - events: list[dict[str, Any]] = [] - if output: - events.append({"type": "tool_output", "content": output}) + exit_code = getattr(item, "exit_code", None) + events: list[dict[str, Any]] = [{ + "type": "tool_output", + "content": output, + "output_complete": True, + **({"output_exit_code": exit_code} if isinstance(exit_code, int) else {}), + }] events.append({"type": "tool_end"}) return events @@ -1463,19 +1470,54 @@ def _file_change_patch_summary(changes: list[Any]) -> str: return "\n".join(lines) -def _is_closed_turn_error(exc: BaseException) -> bool: - """Returns True when the live turn handle is already closed/dead.""" +def _is_transport_death(exc: BaseException) -> bool: + """Returns True when the app-server connection itself died. + + Strictly the transport: the SDK's own TransportClosedError, or an RPC + error the app-server raised about a closed/dead channel. Deliberately + excludes plain RuntimeErrors, because the runner reraises every + non-retryable provider ErrorNotification as `RuntimeError(message)` — + an MCP server "is not running" is a provider fault to report, not a + dead pipe. + + The transport class comes from `_sdk_imports()` and is matched with + isinstance, not by class name: this predicate decides whether an error + reaches the owner at all, so it must be bound to the real symbol (and + must match its subclasses) rather than to anything that happens to + share a name. + """ sdk: dict[str, Any] | None = None try: sdk = _sdk_imports() - except ModuleNotFoundError: + except ImportError: + # ImportError, not ModuleNotFoundError: an SDK that renames or drops + # TransportClosedError fails the `from ... import` the same way a missing + # package does, and this predicate runs INSIDE the turn's except handler — + # raising here would mask the very exception it was asked to classify. sdk = None - if sdk is not None and isinstance( - exc, (sdk["InvalidParamsError"], sdk["CodexRpcError"]) - ): + if sdk is None: + # Without the SDK no turn can have started, so no transport of ours can + # have died. Matching on a class name alone would be worse than useless + # here: it would let any look-alike be mistaken for the real thing. + return False + if isinstance(exc, sdk["TransportClosedError"]): + return True + # InvalidParamsError is a subclass of CodexRpcError, so matching the base + # class alone already covers it. + if isinstance(exc, sdk["CodexRpcError"]): text = str(exc).lower() return "closed" in text or "not running" in text or "broken pipe" in text - if exc.__class__.__name__ == "TransportClosedError": + return False + + +def _is_closed_turn_error(exc: BaseException) -> bool: + """Returns True when the live turn handle is already closed/dead. + + Wider than `_is_transport_death`: a steer against a finished turn also + surfaces as a plain RuntimeError, and there the cost of a false positive + is only a refused steer. + """ + if _is_transport_death(exc): return True if isinstance(exc, RuntimeError): text = str(exc).lower() @@ -1917,6 +1959,33 @@ async def run_codex_sdk_turn( def abort_requested() -> bool: return bool(should_abort and should_abort()) + def stop_requested() -> bool: + """True when Möbius, not the provider, ended this turn. + + The single definition of "we did this to ourselves", shared by terminal + validation (which sees a clean TurnStatus.interrupted) and the except + path (which sees the transport die mid-stream because force_stop killed + the turn's process group). Both need the same fact. + + Includes the superseded-generation abort: a newer turn taking over this + chat is still Möbius ending this one. That leg can be true before + `active_turn` exists, which is deliberate — a teardown during startup is + no more the provider's fault than one mid-stream. + """ + return bool( + (active_turn is not None and active_turn.interrupt_requested) + or abort_requested() + ) + + def with_usage(result: RunnerResult) -> RunnerResult: + """Attaches whatever the turn spent before it ended, however it ended.""" + if final_token_usage is not None: + result["usage"] = _model_dump(final_token_usage) + result["usage_metrics"] = normalize_codex_usage( + first_token_usage, final_token_usage, + ) + return result + def aborted_result() -> RunnerResult: return { "session_id": current_session_id, @@ -2295,33 +2364,49 @@ def aborted_result() -> RunnerResult: error_text, terminal_status, final_message_phase = _codex_terminal_error( completed_turn, sdk, - interrupt_requested=bool( - (active_turn and active_turn.interrupt_requested) - or abort_requested() - ), + interrupt_requested=stop_requested(), completed_message_phases=completed_message_phases, ) - result: RunnerResult = { + result: RunnerResult = with_usage({ "session_id": current_session_id, "cost_usd": None, "error": error_text, - } - if final_token_usage is not None: - result["usage"] = _model_dump(final_token_usage) - result["usage_metrics"] = normalize_codex_usage( - first_token_usage, final_token_usage, - ) + }) if terminal_status is not None: result["terminal_status"] = terminal_status if final_message_phase is not None: result["final_message_phase"] = final_message_phase return result except Exception as exc: - return { + if _is_transport_death(exc) and stop_requested(): + # Our own teardown, seen from the inside. A stop interrupts the turn; + # when that times out the escalation SIGTERMs the turn's private + # process group, so the transport dies mid-stream instead of + # delivering turn/completed. Surfacing that as a provider failure is + # both wrong and destructive: the raw string overwrites the stall note + # (chat._pause_note) published moments earlier, because error blocks + # coalesce latest-wins and drop every events.ERROR_PASSTHROUGH_FIELDS + # the new event omits — taking the note's one-tap Resume with it and + # leaving the owner an unexplained error and no way back. + # + # Usually expected after escalation, but WARNING is deliberate: a real + # app-server crash can coincide with a requested stop and has the same + # transport shape. The dying words are the only forensic evidence left + # once the owner-facing error is suppressed. + log.warning( + "Codex transport closed by our own stop chat_id=%s: %s", chat_id, exc, + ) + return with_usage({ + "session_id": current_session_id, + "cost_usd": None, + "error": None, + "terminal_status": _enum_wire_value(sdk["TurnStatus"].interrupted), + }) + return with_usage({ "session_id": current_session_id, "cost_usd": None, "error": str(exc), - } + }) finally: deferred_cancel: asyncio.CancelledError | None = None if process_group_capture_stop is not None: diff --git a/backend/app/events.py b/backend/app/events.py index 2aee549d7..74900765f 100644 --- a/backend/app/events.py +++ b/backend/app/events.py @@ -334,6 +334,20 @@ def _tool_output_exit_code(content: str, parsed): return int(m.group(1)) if m else None +def tool_output_exit_code(content: object): + """Return a typed command exit code from any sized tool output.""" + if not isinstance(content, str): + return None + parsed = None + stripped = content.lstrip() + if stripped[:1] in ("{", "["): + try: + parsed = json.loads(content) + except (ValueError, TypeError): + parsed = None + return _tool_output_exit_code(content, parsed) + + def excerpt_tool_output(content: str): """Reduce a large tool_output string to (excerpt, full_len, exit_code). @@ -574,6 +588,8 @@ def process_event(event: dict, assistant_blocks: list) -> bool: tool_use_id = event.get("tool_use_id") if tool_use_id: block["tool_use_id"] = tool_use_id + if isinstance(event.get("recall"), dict): + block["recall"] = event["recall"] assistant_blocks.append(block) return True @@ -581,12 +597,21 @@ def process_event(event: dict, assistant_blocks: list) -> bool: # Backfill the input summary. Prefer an exact tool_use_id match (Codex # backfills a WebSearch query at completion, when several searches may be # in flight); older id-less events retain the earliest input-less fallback. + def _apply_tool_input(blk: dict) -> None: + blk["input"] = event.get("input", "") + # A Memory lookup is identified from the command it runs (stamped on the + # event by the sink). Carrying that marker onto the block is what lets the + # turn name the recall while it is still running, and is the ONLY thing + # that authorizes the later output phase to cite notes. + if isinstance(event.get("recall"), dict): + blk["recall"] = event["recall"] + tool_use_id = event.get("tool_use_id") if tool_use_id: for blk in assistant_blocks: if (blk.get("type") == "tool" and blk.get("tool_use_id") == tool_use_id): - blk["input"] = event.get("input", "") + _apply_tool_input(blk) return True candidates = [ blk for blk in assistant_blocks @@ -596,12 +621,12 @@ def process_event(event: dict, assistant_blocks: list) -> bool: ] if len(candidates) == 1: candidates[0]["tool_use_id"] = tool_use_id - candidates[0]["input"] = event.get("input", "") + _apply_tool_input(candidates[0]) return True return False for blk in assistant_blocks: if blk.get("type") == "tool" and not blk.get("input"): - blk["input"] = event.get("input", "") + _apply_tool_input(blk) break return True @@ -615,12 +640,17 @@ def process_event(event: dict, assistant_blocks: list) -> bool: # the full text and read a failure exit code from a field, not a parse # of the possibly-carved excerpt. Absent fields leave the block shape # unchanged (a small, un-reduced output). + exit_code = event.get("output_exit_code") + if exit_code is not None: + blk["output_exit_code"] = exit_code if event.get("output_truncated"): blk["output_truncated"] = True blk["output_full_len"] = event.get("output_full_len") - exit_code = event.get("output_exit_code") - if exit_code is not None: - blk["output_exit_code"] = exit_code + # Settle a Memory lookup from "searching" to what it actually recalled. + # The app prints its bounded result last, so it survives any head+tail + # carving the sink performed before parsing. + if isinstance(event.get("recall"), dict): + blk["recall"] = event["recall"] return True return False diff --git a/backend/app/memory_recall.py b/backend/app/memory_recall.py new file mode 100644 index 000000000..62f450ad4 --- /dev/null +++ b/backend/app/memory_recall.py @@ -0,0 +1,270 @@ +"""Recognize Memory-app recall lookups so a turn can cite what it remembered. + +The Memory app is an ordinary installed app: the agent consults it by running +``memory_search.py`` through Bash, and the notes it read come back as ordinary +tool output. Without this module that lookup is indistinguishable from any +other shell command, so the owner cannot tell "it remembered something" from +"it ran housekeeping" — nor, more importantly, "it looked and found nothing" +from "it never looked". + +Detection is deliberately two-phase and keyed off the tool's own lifecycle: + +* ``recall_from_command`` accepts only the simple absolute invocation documented + by the Memory skill. It deliberately rejects shell composition rather than + trying to partially parse Bash. +* ``recall_from_result`` reads the Memory app's bounded structured result line + and is only called for a tool already identified by the first phase. + +The structured line is printed last, so head+tail carving preserves it. Human +prose and the legacy ``FILES:`` line remain useful to the agent, but neither is +parsed for product state. Missing, malformed, or contradictory result metadata +is an explicit failed lookup, never a successful note-less recall. +""" + +from __future__ import annotations + +import json +import re +import shlex + +# Recall metadata rides inline on the SSE event, the persisted tool block, and +# the compacted activity summary — the same budget the web-source citations +# live within. memory_search.py itself returns at most 4 files with 900-char +# excerpts; these ceilings leave headroom without letting a malformed or +# hostile stdout inflate every transcript read. +MAX_RECALL_NOTES = 12 +MAX_RECALL_TITLE_CHARS = 120 +MAX_RECALL_EXCERPT_CHARS = 300 +MAX_RECALL_PATH_CHARS = 256 +_MAX_OUTPUT_SCAN_CHARS = 262_144 +_MAX_SECTION_LINES_SCANNED = 256 + +RECALL_SEARCHING = "searching" +RECALL_HIT = "hit" +RECALL_EMPTY = "empty" +RECALL_FAILED = "failed" + +# The command summary is the verbatim Bash command. Accept only Memory's +# documented simple absolute invocation. This is intentionally a narrow +# protocol, not a growing shell grammar: composition, redirection, substitution, +# and relative scripts all yield no observability marker while the command +# itself continues to run normally. +_MAX_COMMAND_SCAN_CHARS = 8192 +_ENV_ASSIGN_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=") +_INTERPRETER_RE = re.compile(r"^(?:.*/)?python[0-9.]*$") +_SCRIPT_RE = re.compile( + r"^/data/apps/(?Pmemory(?:-[0-9]+)?)/memory_search\.py$" +) +_CONTROL_TOKEN_RE = re.compile(r"^[;&|()<>]+$") + +_RESULT_PREFIX = "MOBIUS_MEMORY_RESULT_V1:" +_RESULT_RE = re.compile( + rf"^{re.escape(_RESULT_PREFIX)}(?P\{{.*\}})[ \t]*$", + re.MULTILINE, +) + +# A citation path is only ever a repository-relative markdown pointer. Refusing +# anything else keeps traversal, absolute paths, and control characters out of +# a value the client turns into a deep link. +_PATH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*\.md$") +_NOTE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") +_MAX_NOTE_ID_CHARS = 128 + + +def _clean(value: str, limit: int) -> str: + """Collapse whitespace and bound a label taken from tool output.""" + if not isinstance(value, str): + return "" + # Slice before normalizing so a pathological line cannot allocate another + # full-size string merely to produce a short label. + return re.sub(r"\s+", " ", value[: limit * 2]).strip()[:limit] + + +def _note_id(path: str) -> str: + """The graph node id for a citation path: its file stem.""" + tail = path.rsplit("/", 1)[-1] + return tail[:-3] if tail.endswith(".md") else tail + + +def _safe_note_id(value: object, path: str) -> str: + """Keep the graph's real node id, with a path-stem fallback for old apps. + + A graph id is not required to equal its markdown filename. The Memory app + opens nodes by id, so replacing a valid structured id with the path stem + makes a well-formed citation navigate to nowhere whenever those differ. + """ + if isinstance(value, str): + candidate = value.strip() + if ( + candidate + and len(candidate) <= _MAX_NOTE_ID_CHARS + and _NOTE_ID_RE.fullmatch(candidate) + ): + return candidate + return _note_id(path) + + +def _title_from_path(path: str) -> str: + """A readable fallback when the titled section line was carved away.""" + return _note_id(path).replace("-", " ").replace("_", " ").strip() + + +def _safe_path(value: str) -> str: + if not isinstance(value, str): + return "" + candidate = value.strip() + if not candidate or len(candidate) > MAX_RECALL_PATH_CHARS: + return "" + if ".." in candidate or candidate.startswith("/"): + return "" + return candidate if _PATH_RE.match(candidate) else "" + + +def _simple_command_tokens(command: str) -> list[str] | None: + try: + lexer = shlex.shlex(command, posix=True, punctuation_chars=True) + lexer.whitespace_split = True + lexer.commenters = "" + tokens = list(lexer) + except ValueError: + return None + if not tokens or any(_CONTROL_TOKEN_RE.fullmatch(token) for token in tokens): + return None + # Reject substitutions/backticks conservatively. They are not part of the + # documented call and would turn this recognizer back into a shell parser. + if any("`" in token or "$(" in token for token in tokens): + return None + return tokens + + +def _tokens_search_slug(tokens: list[str]) -> str | None: + """Return the invoked Memory app slug for the exact documented command. + + Exact arity is a security boundary, not mere tidiness: ``shlex`` treats a + newline as whitespace, so accepting arbitrary trailing tokens would also + accept a second shell command whose output could forge the structured result + line. The supported command is env assignments + Python flags + script + + query + chat id, and then it must end. + """ + index = 0 + while index < len(tokens) and _ENV_ASSIGN_RE.match(tokens[index]): + index += 1 + if index >= len(tokens): + return None + head = tokens[index] + direct = _SCRIPT_RE.fullmatch(head) + if direct: + return direct.group("app_slug") if len(tokens) == index + 3 else None + if not _INTERPRETER_RE.match(head): + return None + for script_index, token in enumerate(tokens[index + 1:], start=index + 1): + if token.startswith("-"): + continue + script = _SCRIPT_RE.fullmatch(token) + if script and len(tokens) == script_index + 3: + return script.group("app_slug") + return None + return None + + +def recall_from_command(command: object) -> dict | None: + """Return a pending recall marker when this command RUNS a memory lookup. + + Called at tool-input time so the live turn can name what it is doing while + the lookup is still in flight. Returning ``None`` means "not a memory + lookup", which is also the safe answer for a missing or oversized command + summary — and, deliberately, for any command that merely names the script. + """ + if not isinstance(command, str) or not command: + return None + if len(command) > _MAX_COMMAND_SCAN_CHARS: + return None + # Cheap reject before tokenizing: the overwhelming majority of commands are + # not memory lookups and should cost one substring scan. + if "memory_search.py" not in command: + return None + tokens = _simple_command_tokens(command) + app_slug = _tokens_search_slug(tokens) if tokens else None + return ( + {"status": RECALL_SEARCHING, "app_slug": app_slug} + if app_slug else None + ) + + +def recall_from_result(text: object, exit_code: object = None) -> dict: + """Validate a known Memory command's final structured result.""" + if isinstance(exit_code, bool): + exit_code = None + if isinstance(exit_code, int) and exit_code != 0: + return {"status": RECALL_FAILED} + if not isinstance(text, str) or not text.strip(): + return {"status": RECALL_FAILED} + + body = text[-_MAX_OUTPUT_SCAN_CHARS:] + matches = list(_RESULT_RE.finditer(body)) + if not matches: + return {"status": RECALL_FAILED} + try: + payload = json.loads(matches[-1].group("payload")) + except (TypeError, ValueError, json.JSONDecodeError): + return {"status": RECALL_FAILED} + if not isinstance(payload, dict): + return {"status": RECALL_FAILED} + + status = payload.get("status") + if status == RECALL_FAILED: + return {"status": RECALL_FAILED} + if status == RECALL_EMPTY: + return {"status": RECALL_EMPTY} + if status != RECALL_HIT or not isinstance(payload.get("notes"), list): + return {"status": RECALL_FAILED} + + notes: list[dict[str, str]] = [] + seen: set[str] = set() + for raw_note in payload["notes"][:_MAX_SECTION_LINES_SCANNED]: + if not isinstance(raw_note, dict): + continue + path = _safe_path(raw_note.get("path")) + if not path or path in seen: + continue + seen.add(path) + title = _clean(raw_note.get("title"), MAX_RECALL_TITLE_CHARS) + excerpt = _clean(raw_note.get("excerpt"), MAX_RECALL_EXCERPT_CHARS) + note = { + "id": _safe_note_id(raw_note.get("id"), path), + "path": path, + "title": title or _title_from_path(path) or path, + } + if excerpt: + note["excerpt"] = excerpt + notes.append(note) + if len(notes) >= MAX_RECALL_NOTES: + break + return ( + {"status": RECALL_HIT, "notes": notes} + if notes else {"status": RECALL_FAILED} + ) + + +def merge_recall_notes( + target: list[dict[str, str]], + seen: set[str], + recall: object, +) -> None: + """Accumulate one block's notes into a deduped, bounded citation list. + + Shared by the transcript compaction rollup so the projection and the live + block agree on ordering (first occurrence owns the position) and on the cap. + """ + if not isinstance(recall, dict): + return + for note in recall.get("notes") or []: + if not isinstance(note, dict): + continue + path = note.get("path") + if not isinstance(path, str) or not path or path in seen: + continue + seen.add(path) + target.append(note) + if len(target) >= MAX_RECALL_NOTES: + return diff --git a/backend/app/platform_update.py b/backend/app/platform_update.py index 70fd4a785..bacad3bc0 100644 --- a/backend/app/platform_update.py +++ b/backend/app/platform_update.py @@ -1,12 +1,12 @@ -"""Platform self-update — clone-native ``git fetch`` + rebase reconcile. +"""Platform self-update — clone-native ``git fetch`` + merge reconcile. ``/data/platform`` is a real ``git clone`` of the canonical repo; uvicorn serves its backend directly (``cd /data/platform/backend && uvicorn app.main:app``). Local ``main`` carries the agent's edits; the ``upstream`` branch records the commit the clone was last reconciled to (set to HEAD at clone time). A deploy ships a new image AND advances canonical ``origin/main``; this module makes that -deploy actually REACH a running instance by fetching origin and replaying the -local edits onto the new upstream — on boot (before uvicorn imports the code, so +deploy actually REACH a running instance by fetching origin and merging it with +the local edits — on boot (before uvicorn imports the code, so the update goes live automatically) and on owner-triggered Apply. Owner Apply pins the exact target returned by the review plan even if its fetch observes a newer remote head; backend changes then need a restart to load. @@ -14,18 +14,18 @@ The reconcile is built to be non-destructive above all else: 1. ``/data/platform`` holds the SERVED backend, so a reconcile must never leave a - half-applied tree. A rebase conflict is aborted back to the pre-reconcile + half-applied tree. A merge conflict is aborted back to the pre-reconcile commit (the old, working code keeps serving) and surfaced as a conflict; a - crash mid-rebase is detected on the next boot (``.git/rebase-merge``) and - aborted before anything else runs. + crash mid-merge is detected on the next boot and aborted before anything + else runs. Legacy interrupted rebases are still cleaned up too. 2. Local edits are NEVER lost. Uncommitted working-tree edits are committed onto - ``main`` before any fast-forward/rebase, so a fast-forward ``reset --hard`` or - a rebase can only ever replay them, never discard them. A conflict or an + ``main`` before any fast-forward/merge, so either operation starts from a + durable local tip. A conflict or an import-broken result rolls the served tree back to exactly those local edits. -3. A text-clean rebase can still produce a tree that fails to import (e.g. - upstream deleted a module a local edit still imports). A post-rebase import +3. A text-clean merge can still produce a tree that fails to import (e.g. + upstream deleted a module a local edit still imports). A post-merge import probe catches that and rolls back to the previous served commit rather than serving a broken tree. @@ -35,7 +35,9 @@ module reuses ``app_git``'s isolated git env and ``commit_local`` engine; it does NOT carry forward the old baked-floor machinery (recording a baked tree onto ``upstream``), which fought the clone model — a real ``git fetch origin`` plus a -rebase against real ancestry replaces it entirely. +merge against real ancestry replaces it entirely. Diverged histories merge as +one net change instead of replaying every local commit separately, so all +conflicts surface together and local commit identities remain intact. """ from __future__ import annotations @@ -80,20 +82,20 @@ # on-disk clone after an agent edits /data/platform. SERVING_SOURCE_FILE = Path("/tmp/serving-source") SERVING_SHA_FILE = Path("/tmp/serving-sha") -# Persist a conflict so Settings keeps showing it across reloads (the rebase is +# Persist a conflict so Settings keeps showing it across reloads (the merge is # aborted, so no git state alone can signal it). Records the target sha + paths. CONFLICT_FLAG = Path("/data/.platform-conflict") # Persist that the last reconcile could not refresh origin. Deploy verification # treats this as an explicit exemption from the freshness assertion; the next # successful fetch clears it. OFFLINE_FLAG = Path("/data/.platform-offline") -# A text-clean rebase whose result failed the import probe was rolled back to the +# A text-clean merge whose result failed the import probe was rolled back to the # previous served commit. Records the target sha + the import error so Settings # can show "rolled back — needs repair" rather than silently staying "up to # date". ROLLED_BACK_FLAG = Path("/data/.platform-rolled-back") # Transient crash-safety marker written immediately before reconcile mutates the -# served tree. If the boot subprocess is SIGKILLed mid-rebase/probe/rollback, the +# served tree. If the boot subprocess is SIGKILLed mid-merge/probe/rollback, the # post-timeout boot guard uses this sha to restore the last committed served tip # before uvicorn imports anything. RECONCILE_PRE_FLAG = Path("/data/.platform-reconcile-pre") @@ -116,7 +118,12 @@ # this is wedged, not busy. Fetch gets its own (network-bound) budget. _GIT_TIMEOUT = 120 _FETCH_TIMEOUT = 120 -# The post-rebase import probe. A module-level infinite loop or a blocking call +# Distinct sentinel returned by ``_merge_target`` when the merge wedged and was +# aborted (as opposed to a positive git returncode from a content conflict). +# After the abort no unmerged paths survive, so the caller must treat this as an +# error/serve-old outcome rather than fabricating a zero-path content conflict. +_MERGE_TIMEOUT = -1 +# The post-merge import probe. A module-level infinite loop or a blocking call # in agent-edited code would otherwise wedge boot forever; a timeout-kill counts # as probe-fail -> roll back. _PROBE_TIMEOUT = 60 @@ -195,7 +202,7 @@ class PlatformUpdateState(str, Enum): AVAILABLE = "available" CONFLICT = "conflict" RESTART_NEEDED = "restart_needed" - # A text-clean rebase failed the import probe and was rolled back to the + # A text-clean merge failed the import probe and was rolled back to the # previous served commit; the update needs a repair pass before it can land. ROLLED_BACK = "rolled_back" @@ -210,7 +217,7 @@ class PlatformStatus(TypedDict): recorded_upstream_sha: str | None # Latest fetched origin/main commit that is already contained in local main. # Unlike recorded_upstream_sha, this remains correct after a manual/agent - # rebase that did not run the updater's marker-maintenance path. + # merge that did not run the updater's marker-maintenance path. contained_upstream_sha: str | None seed_required: bool conflict_paths: list[str] @@ -293,9 +300,9 @@ class ReconcileResult: """Outcome of a single :func:`reconcile_clone` pass. ``status`` is one of ``up_to_date`` (origin already integrated), ``updated`` - (fast-forward or rebase applied and the import probe passed), ``conflict`` - (rebase conflicted, aborted, serving the pre sha), ``rolled_back`` (text-clean - rebase failed the import probe, reset to the pre sha), ``offline`` (fetch + (fast-forward or merge applied and the import probe passed), ``conflict`` + (merge conflicted, aborted, serving the pre sha), ``rolled_back`` (text-clean + merge failed the import probe, reset to the pre sha), ``offline`` (fetch failed — kept serving unchanged), ``skipped`` (not a reconcilable clone), or ``error`` (an unexpected git failure was caught and the served tree reset to the pre sha). @@ -312,7 +319,7 @@ class ReconcileResult: error: str | None = None # Exact reviewed release/upstream commit captured while RECONCILE_LOCK is # still held. Hook refresh reads every allowlisted blob from this immutable - # generation rather than trusting replayed local HEAD or a moving ref. + # generation rather than trusting a locally merged HEAD or a moving ref. hook_source_sha: str | None = None @@ -427,7 +434,7 @@ def _git( ) -> subprocess.CompletedProcess: """Run ``git -C repo `` in text mode under the scrubbed, ceiling-pinned env. ``check=False`` lets callers read a non-zero return (a merge-base miss, a - rebase conflict) instead of raising.""" + merge conflict) instead of raising.""" return subprocess.run( ["git", "-C", str(repo), *args], capture_output=True, text=True, timeout=timeout, check=check, @@ -470,7 +477,7 @@ def _head_detached(repo: Path = PLATFORM_REPO) -> bool: def _reattach_detached_head(repo: Path, local: str) -> None: """Move the working branch to the current detached HEAD, preserving the worktree. This makes the subsequent ``commit_local`` land on the branch the - reconcile will actually fast-forward/rebase.""" + reconcile will actually fast-forward/merge.""" if _head_detached(repo): _git("checkout", "-B", local, "HEAD", repo=repo) @@ -498,18 +505,24 @@ def _unmerged_paths(repo: Path = PLATFORM_REPO) -> list[str]: def _rebase_in_progress(repo: Path = PLATFORM_REPO) -> bool: + """Legacy sequencer state left by an updater or resolver from an older build.""" git_dir = repo / ".git" return (git_dir / "rebase-merge").exists() or (git_dir / "rebase-apply").exists() +def _merge_in_progress(repo: Path = PLATFORM_REPO) -> bool: + return bool(_rev(repo, "MERGE_HEAD")) + + +def _reconcile_in_progress(repo: Path = PLATFORM_REPO) -> bool: + return _rebase_in_progress(repo) or _merge_in_progress(repo) + + def _abort_interrupted(repo: Path = PLATFORM_REPO) -> None: - """Crash-safety: abort a rebase/merge left half-finished by a prior crash so - the reconcile starts from a clean, committed ``main`` (the pre-crash tip). A - mid-rebase SIGKILL leaves ``.git/rebase-merge``; a stray merge leaves - ``MERGE_HEAD``. Aborting each restores the branch to its state before the op.""" + """Abort a current merge or a legacy rebase left half-finished by a crash.""" if _rebase_in_progress(repo): _git("rebase", "--abort", repo=repo, check=False) - if (repo / ".git" / "MERGE_HEAD").exists(): + if _merge_in_progress(repo): _git("merge", "--abort", repo=repo, check=False) @@ -542,7 +555,7 @@ def boot_guard_clean_served_tree(repo: Path = PLATFORM_REPO) -> str: return "boot_guard[skipped] no_git" local = _local_branch(repo) pre = _read_reconcile_pre() - interrupted = _rebase_in_progress(repo) or (repo / ".git" / "MERGE_HEAD").exists() + interrupted = _reconcile_in_progress(repo) _abort_interrupted(repo) if pre and _rev(repo, pre): _reset_hard_to(repo, local, pre) @@ -569,8 +582,8 @@ def _fetch(repo: Path = PLATFORM_REPO) -> bool: def _fetch_unshallow(repo: Path = PLATFORM_REPO) -> None: - """Deepen a shallow clone so a rebase can find a real merge base. Best-effort: - an offline/timeout failure leaves the clone shallow and the caller's rebase + """Deepen a shallow clone so a merge can find a real merge base. Best-effort: + an offline/timeout failure leaves the clone shallow and the caller's merge either still succeeds (the base was inside the shallow window) or reports a conflict, which fails closed to serve-old — never a hard reset.""" try: @@ -580,38 +593,40 @@ def _fetch_unshallow(repo: Path = PLATFORM_REPO) -> None: pass -def _rebase_onto(repo: Path, target: str, local: str) -> int: - """Rebase the local commits (``main`` beyond the shared base) onto ``target``. - - ``git rebase target local`` replays the commits in ``local`` that are not in - ``target`` on top of ``target`` — i.e. the agent's local edits onto the new - upstream. The Mobius identity is injected per-invocation (``-c user.*``) so a - replay commit never depends on repo/global git config being set — the rebase - writes new commits and would otherwise fail "committer identity unknown" on a - clone with no configured user. The editor is disabled so a replay never blocks - on an interactive editor, and the whole op is bounded by a timeout. Returns the - git return code (0 clean, non-zero on conflict/error).""" - env = { - **_scrubbed_git_env(repo), - "GIT_EDITOR": "true", - "GIT_SEQUENCE_EDITOR": "true", - } +def _merge_target(repo: Path, target: str) -> int: + """Merge one reviewed upstream target into the checked-out local branch. + + A single merge compares the net local and upstream trees from their shared + base. Unlike a rebase, it neither rewrites every local commit nor makes the + resolver discover conflicts one historical commit at a time. ``--no-ff`` is + deliberate: this helper is called only for diverged histories, and the merge + commit preserves both the reviewed upstream target and the complete local + history as explicit parents. Returns 0 on a clean committed merge and a + positive returncode on a content conflict/error; returns the distinct + sentinel :data:`_MERGE_TIMEOUT` when the merge wedged and was aborted (no + unmerged paths survive the abort, so the caller must classify it as an error/ + serve-old outcome rather than a content conflict). The caller owns abort + + serve-old recovery. + """ + env = _scrubbed_git_env(repo) try: proc = subprocess.run( [ "git", "-c", f"user.name={app_git._GIT_NAME}", "-c", f"user.email={app_git._GIT_EMAIL}", - "-C", str(repo), "rebase", target, local, + "-C", str(repo), "merge", "--no-ff", "-m", + f"platform: merge upstream {target[:12]}", target, ], capture_output=True, text=True, timeout=_GIT_TIMEOUT, check=False, env=env, ) return proc.returncode except subprocess.TimeoutExpired: - # A wedged rebase must not leave a half-rebased tree: abort so the caller's - # serve-old path is honoured. - _git("rebase", "--abort", repo=repo, check=False) - return 1 + # A wedged merge must not leave a half-merged tree: abort so the caller's + # serve-old path is honoured. Return a distinct sentinel so the caller does + # not read the (now empty) unmerged paths and fabricate a zero-path conflict. + _git("merge", "--abort", repo=repo, check=False) + return _MERGE_TIMEOUT def _reset_hard_to(repo: Path, local: str, sha: str) -> None: @@ -639,7 +654,7 @@ def _clear_upstream(repo: Path) -> None: def _import_probe(repo: Path = PLATFORM_REPO, timeout: int = _PROBE_TIMEOUT): """Run ``import app.main`` as a fresh subprocess with cwd the served backend. - Single-source probe for both boot and post-rebase: it MUST be a subprocess (not + Single-source probe for both boot and post-merge: it MUST be a subprocess (not an in-process import) so the reconcile process — which already imported the OLD ``app.platform_update`` — validates the NEW on-disk tree without corrupting its own interpreter, and so cwd/env exactly mirror the uvicorn exec. The env scrubs @@ -1087,7 +1102,7 @@ def _rebuild_frontend_after_update_if_needed( ) -> None: """Rebuild served frontend assets after a clean update that changed them. - The live edit watcher sees ordinary file saves, but git checkout/rebase during + The live edit watcher sees ordinary file saves, but git checkout/merge during the Settings update flow can move frontend files without a reliable watcher event. Without this explicit rebuild, ``/data/platform/frontend/src`` advances while ``dist`` keeps serving the old bundle. @@ -1163,7 +1178,7 @@ def reconcile_clone( return ReconcileResult("skipped", None, None, None, error="no_git") local = _local_branch(repo) - # Crash-safety FIRST: a mid-rebase crash must be aborted before anything reads + # Crash-safety FIRST: a mid-reconcile crash must be aborted before anything reads # the tree, so we reconcile from the committed pre-crash tip. _abort_interrupted(repo) pre = _rev(repo, local) @@ -1207,7 +1222,7 @@ def reconcile_clone( if progress: progress(PlatformUpdatePhase.RECONCILING) # A deploy advanced origin beyond committed main. Commit any uncommitted edits - # FIRST so neither the fast-forward reset nor the rebase can discard them. + # FIRST so neither the fast-forward reset nor the merge can discard them. _reattach_detached_head(repo, local) app_git.commit_local(repo, "platform: local edits before reconcile") pre = _rev(repo, local) # now includes the just-committed edits @@ -1224,7 +1239,7 @@ def reconcile_clone( # behind. The normal fetch has already transferred the new first-parent # chain, so Git can prove the overwhelmingly common fast-forward directly. # Only a shallow clone whose ancestry is still ambiguous needs the expensive - # full-history fallback before we choose between reset and rebase. + # full-history fallback before we choose between reset and merge. fast_forward = bool(pre) and _is_ancestor(repo, pre, target) if _is_shallow(repo) and not fast_forward: if progress: @@ -1241,22 +1256,36 @@ def reconcile_clone( # discard committed local edits. _git("reset", "--hard", target, repo=repo) else: - # main has commits not in target (diverged): REBASE local edits onto the new - # upstream so BOTH survive. - rc = _rebase_onto(repo, target, local) + # Main and target diverged: merge the reviewed upstream tree ONCE. This + # preserves local commit identities and makes one resolver pass see every + # net conflict instead of stopping once per historical local commit. + rc = _merge_target(repo, target) if rc != 0: - # Conflict: NEVER leave a half-rebased tree. Abort back to PRE (the old, - # working code keeps serving), record the conflict, clear any stale - # rollback flag, and let the caller open a resolver chat. + # NEVER leave a half-merged tree. Read the unmerged paths BEFORE the + # abort clears them. paths = _unmerged_paths(repo) - _git("rebase", "--abort", repo=repo, check=False) + _git("merge", "--abort", repo=repo, check=False) _reset_hard_to(repo, local, pre) # belt-and-braces: ensure main == PRE + # A wedged merge (timeout sentinel) — or, defensively, ANY nonzero + # result that produced no unmerged paths — is NOT a reviewable content + # conflict: ``_merge_target`` already aborted it, so there is nothing for + # a resolver chat to reconcile. Classify it as an error/serve-old outcome + # (reset to PRE, no conflict flag, no resolver chat) rather than + # fabricating a zero-path conflict. + if rc == _MERGE_TIMEOUT or not paths: + CONFLICT_FLAG.unlink(missing_ok=True) + ROLLED_BACK_FLAG.unlink(missing_ok=True) + _clear_reconcile_pre() + err = "merge_timeout" if rc == _MERGE_TIMEOUT else "merge_failed" + return ReconcileResult("error", pre, pre, target, error=err) + # Content conflict: record it, clear any stale rollback flag, and let the + # caller open a resolver chat. _write_conflict_flag(target, paths) ROLLED_BACK_FLAG.unlink(missing_ok=True) _clear_reconcile_pre() return ReconcileResult("conflict", pre, pre, target, conflict_paths=paths) - # Post-reconcile import probe: a text-clean ff/rebase can still produce a + # Post-reconcile import probe: a text-clean ff/merge can still produce a # tree that fails to import (upstream dropped a module a local edit imports; # a bad deploy). Roll back to the previous served commit rather than serve it # broken. Skip the ~60s throwaway boot when the reconcile touched NO served @@ -1280,7 +1309,7 @@ def reconcile_clone( _clear_reconcile_pre() return ReconcileResult("error", pre, pre, target, error=repr(exc)) - # Success: main now carries the update plus any replayed local edits. Advance + # Success: main now carries the update plus all local edits. Advance # the upstream marker and clear conflict/rollback flags. At boot the fresh # uvicorn imports this directly (clear the restart flag — the boot IS the # restart the flag would ask for); an owner Apply marks a restart via the @@ -1330,7 +1359,7 @@ def _reconcile_under_lock( at_boot=at_boot, # A reviewed Apply already proved the immutable object exists. Fetching a # moving remote again adds latency and was the original TOCTOU bug; boot - # keeps the normal refresh path. A shallow rebase may still deepen below. + # keeps the normal refresh path. A shallow merge may still deepen below. fetch_remote=plan_id is None, progress=progress, ) @@ -1354,7 +1383,7 @@ def _reconcile_under_lock( ) # `upstream` is moved only by a successful/contained reconcile to the # fetched release target. Capture its immutable oid before releasing the - # cross-process lock; local replay commits on main are intentionally not a + # cross-process lock; local commits on main are intentionally not a # hook trust transition. return replace( result, @@ -1416,7 +1445,7 @@ def platform_status(repo: Path = PLATFORM_REPO) -> PlatformStatus: """ image_sha = current_build_sha() upstream_sha = recorded_upstream_sha(repo) - conflict = CONFLICT_FLAG.exists() or _rebase_in_progress(repo) + conflict = CONFLICT_FLAG.exists() or _reconcile_in_progress(repo) rolled_back = ROLLED_BACK_FLAG.exists() restart_needed = RESTART_NEEDED_FLAG.exists() or _platform_tree_needs_restart(repo) local = _local_branch(repo) @@ -1777,7 +1806,7 @@ async def create_platform_conflict_resolver_chat( from app import models flag = _read_conflict_flag() or {} - if not (CONFLICT_FLAG.exists() or _rebase_in_progress(repo)): + if not (CONFLICT_FLAG.exists() or _reconcile_in_progress(repo)): raise PlatformUpdateError("No unresolved platform update conflict.") existing_chat_id = flag.get("chat_id") @@ -1795,24 +1824,56 @@ async def create_platform_conflict_resolver_chat( ) conflict_paths = flag.get("paths") or _unmerged_paths(repo) - result = await spawn_platform_conflict_chat(db, conflict_paths) + target_sha = flag.get("upstream") or _rev(repo, DEFAULT_TARGET_REF) + if not target_sha: + raise PlatformUpdateError("Platform conflict target is unavailable.") + result = await spawn_platform_conflict_chat(db, conflict_paths, target_sha) if result is None: raise PlatformUpdateError("Could not open resolver chat.") _write_conflict_flag( - flag.get("upstream") or _rev(repo, DEFAULT_TARGET_REF), + target_sha, conflict_paths, result["chat_id"], ) return result +def _platform_conflict_resolver_message( + target_sha: str, + conflict_paths: list[str], +) -> str: + """Instructions bound to the exact release the owner reviewed and applied.""" + files = ", ".join(conflict_paths) if conflict_paths else "some files" + return ( + "A platform update is ready but conflicts with local edits — the new " + "version and the local changes both touched the same lines, so they can't " + "merge cleanly.\n\n" + "The clone at `/data/platform` is a real git checkout of the platform repo. " + f"The exact reviewed version is commit `{target_sha}`; local edits are on " + "the checked-out working branch. " + f"Reconcile these conflicting files by hand: {files}.\n\n" + "Resolve it with ordinary git: `git -C /data/platform merge --no-ff " + f"{target_sha}` compares the complete local and reviewed upstream trees " + "once and stops with every conflicting file marked; combine the intent of " + "the local version and upstream's, save each file, then `git add` it and " + "`git commit --no-edit` (this finishes the merge non-interactively from the " + "prepared merge message). When the merge finishes, the working branch " + "carries both histories.\n\n" + "When the reconcile is committed, clear the flag " + "(`rm -f /data/.platform-conflict`) and tell the owner to **restart the " + "server** from Settings to finish. To back out instead, `git -C " + "/data/platform merge --abort`, `rm -f /data/.platform-conflict`, and tell " + "the owner the update was skipped." + ) + + async def spawn_platform_conflict_chat( - db: Session, conflict_paths: list[str], + db: Session, conflict_paths: list[str], target_sha: str, ) -> PlatformConflictResolverChatOut | None: """Open a visible agent chat to reconcile the new platform version into - ``main`` — the platform analogue of a per-app update-conflict resolver chat. - Dedupes on a running resolver.""" + the checked-out working branch — the platform analogue of a per-app + update-conflict resolver chat. Dedupes on a running resolver.""" import time import uuid @@ -1845,25 +1906,7 @@ async def spawn_platform_conflict_chat( get_settings().data_dir, owner.provider, ) - files = ", ".join(conflict_paths) if conflict_paths else "some files" - content = ( - "A platform update is ready but conflicts with local edits — the new " - "version and the local changes both touched the same lines, so it can't " - "rebase cleanly.\n\n" - "The clone at `/data/platform` is a real git checkout of the platform repo. " - "The new version is on the fetched `origin/main`; local edits are on `main`. " - f"Reconcile these conflicting files by hand: {files}.\n\n" - "Resolve it with ordinary git: `git -C /data/platform rebase origin/main` " - "replays the local edits onto the new version and stops on the conflicting " - "files with conflict markers; for each, combine the intent of the local " - "version and origin's, save it, then `git add` it and `git rebase " - "--continue`. When the rebase finishes, `main` carries both.\n\n" - "When the reconcile is committed, clear the flag " - "(`rm -f /data/.platform-conflict`) and tell the owner to **restart the " - "server** from Settings to finish. To back out instead, `git -C " - "/data/platform rebase --abort`, `rm -f /data/.platform-conflict`, and tell " - "the owner the update was skipped." - ) + content = _platform_conflict_resolver_message(target_sha, conflict_paths) chat_id = str(uuid.uuid4()) chat = models.Chat( diff --git a/backend/app/routes/chats_stream.py b/backend/app/routes/chats_stream.py index ef645af64..710c0130c 100644 --- a/backend/app/routes/chats_stream.py +++ b/backend/app/routes/chats_stream.py @@ -21,6 +21,7 @@ is_draining, mark_starting, run_chat, + steered_into_turn_event, ) from app import chat_queue from app.chat_writer import ( @@ -409,16 +410,30 @@ async def _split_steer_at_route( def _steered_response( - chat_id: str, pending_messages: list[dict] | None = None, + chat_id: str, + pending_messages: list[dict] | None = None, + *, + cut_deferred: bool = False, ) -> JSONResponse: """202 for a message steered into the live turn. - The message is in the TRANSCRIPT (not the pending queue) and the live - turn already saw it via `steer()`, so the client renders it inline as - content growth rather than a queued-tray entry.""" + The live turn has seen the message via `steer()`. Where the message LIVES + right now depends on the provider, and `cut_deferred` states which: + + - absent/False (Codex): the transcript split already ran at the route, so the + row is in the TRANSCRIPT and out of the pending queue. The client drops its + queued-tray entry and renders the row inline as content growth. + - True (Claude): the split is deferred to the runner's interrupt boundary, so + the row is STILL in `pending_messages` (echoed above) and will move into the + transcript when the runner publishes `steered_into_turn` at the real seal. + The client keeps showing the queued row until then — dropping it here left + the owner's message with nowhere to render for the length of the window. + """ payload = {"status": "steered", "chat_id": chat_id} if pending_messages is not None: payload["pending_messages"] = pending_messages + if cut_deferred: + payload["cut_deferred"] = True return JSONResponse(status_code=202, content=payload) @@ -819,8 +834,10 @@ def _coerce_chat_settings(c): # Stop's queue-collapse path may pass force_steer to turn already # queued messages into a live steer. Codex injects into the running # SDK turn; Claude interrupts and re-prompts on the same client. On - # success the user message goes into the transcript and a - # `steered_into_turn` event tells the client to render it inline. + # success a `steered_into_turn` event tells the client the transcript has + # been split and the user row is in it — published here for Codex (the + # route is its seal point) and by the runner for Claude, whose split waits + # for the interrupt boundary (see the publish below). provider = chat.provider or "claude" if ( is_chat_running(chat_id) @@ -872,14 +889,16 @@ def _coerce_chat_settings(c): steered = False if steered: if defer_to_runner: - # The optimistic response mirrors the runner's cid conversion. - consumed = set(consume_cids) + # Nothing has been converted yet: the rows are buffered on the handle + # and remain in `chat.pending_messages` until the runner's seal + # consumes them. So report the queue AS IT IS — an optimistic list + # with the rows already subtracted told the client they had left the + # queue while the server still held them there, which is precisely + # the window the client must keep showing them (see `cut_deferred` + # on the response below). stored_result = { "stored_messages": user_msgs, - "pending": [ - m for m in (chat.pending_messages or []) - if cid_of(m) not in consumed - ], + "pending": list(chat.pending_messages or []), } else: # Codex append and pending removal commit atomically for one cid. @@ -900,31 +919,28 @@ def _coerce_chat_settings(c): status_code=503, detail="Could not save your message; please refresh.", ) - bc = get_broadcast(chat_id) - if bc is not None: - stored_messages = stored_result.get("stored_messages") - if not isinstance(stored_messages, list) or not stored_messages: - stored = stored_result.get("stored") or user_msg - stored_messages = [stored] - bc.publish({ - "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", ""), - }) + # `steered_into_turn` is the client's ONLY "cut the live stream here" + # signal, so it may only be published where the transcript is really + # split. Codex splits HERE (`_split_steer_at_route` above ran to + # completion), so the route is its seal point and publishes the cut + # unchanged. Claude defers the split to the runner's interrupt boundary, + # seconds later — publishing the cut here re-based the client's stream + # while the runner was still emitting blocks that the deferred seal then + # folded into A1, so those blocks painted twice for the rest of the turn. + # The deferred path publishes NOTHING here: the 202 below carries the + # still-queued row, which is the single signal that keeps it visible + # until `_seal_steer_split` publishes the cut. + if not defer_to_runner: + bc = get_broadcast(chat_id) + if bc is not None: + stored_messages = stored_result.get("stored_messages") + if not isinstance(stored_messages, list) or not stored_messages: + stored = stored_result.get("stored") or user_msg + stored_messages = [stored] + bc.publish(steered_into_turn_event(stored_messages)) return _steered_response( chat_id, stored_result.get("pending"), + cut_deferred=defer_to_runner, ) if body.force_steer: return _not_steered_response(chat_id) diff --git a/backend/scripts/entrypoint.sh b/backend/scripts/entrypoint.sh index 2d246f2fc..8526bf976 100755 --- a/backend/scripts/entrypoint.sh +++ b/backend/scripts/entrypoint.sh @@ -85,7 +85,7 @@ if [ "$_boot_counter" -ge 3 ] && [ -f /data/.last-successful-boot ]; then # TIMESTAMPED /data/platform.crashloop-prev. for inspection/recovery, not # deleted. A one-slot .crashloop-prev would let a SECOND crash-loop delete the # first preserved tree before the owner could inspect it, so we timestamp each - # quarantine and keep only the newest few. (slice B's deploy=rebase + # quarantine and keep only the newest few. (slice B's deploy=merge # reconciliation will refine this.) _cl_ts=$(date -u +%Y%m%dT%H%M%SZ) if [ -e /data/platform ] && [ -n "$(ls -A /data/platform 2>/dev/null)" ] && @@ -736,9 +736,9 @@ printf '%s\n' "$_served_sha" > /tmp/serving-sha chmod 644 /tmp/serving-source /tmp/serving-sha 2>/dev/null || true if [ "$_use_platform" -eq 1 ] && [ "${MOBIUS_TEST_RUNTIME:-0}" != "1" ]; then - # Slice B deploy=rebase reconcile. A deploy ships a new image AND advances - # canonical origin/main; fetch origin and replay the local edits onto the new - # version NOW, before uvicorn imports the code, so the update goes live this + # Slice B deploy=merge reconcile. A deploy ships a new image AND advances + # canonical origin/main; fetch origin and merge the new version once with the + # local edits NOW, before uvicorn imports the code, so the update goes live this # boot with no restart. Runs as mobius (writes /data; root would poison /data # ownership + hit git "dubious ownership"), cwd the served backend so `app` # imports resolve from the clone, under the IDENTICAL GIT_*/PYTHONPATH scrub @@ -746,12 +746,12 @@ if [ "$_use_platform" -eq 1 ] && [ "${MOBIUS_TEST_RUNTIME:-0}" != "1" ]; then # `|| true` guards the shell, so a reconcile failure never bricks boot; a # conflict/rollback leaves the pre-reconcile code on disk (aborted/reset) and # sets a flag Settings surfaces. The outer `timeout` is a last-resort bound set - # ABOVE the reconcile's bounded operations: fetch 120 + unshallow 120 + rebase + # ABOVE the reconcile's bounded operations: fetch 120 + unshallow 120 + merge # 120 + probe 60 = 420, plus commit_local's own bounded git calls. Keep this # comfortably higher so internal timeouts fire FIRST; the post-timeout guard # below still cleans the tree if the outer kill ever wins. recoveryd remains # the outer floor. - echo "Platform layer: reconciling /data/platform with origin (slice B deploy=rebase)..." >&2 + echo "Platform layer: reconciling /data/platform with origin (slice B deploy=merge)..." >&2 su -s /bin/sh mobius -c \ "cd /data/platform/backend && $_env_scrub timeout 900 python3 -c \ 'from app import platform_update; print(platform_update.reconcile_clone_sync())'" \ @@ -767,7 +767,7 @@ if [ "$_use_platform" -eq 1 ] && [ "${MOBIUS_TEST_RUNTIME:-0}" != "1" ]; then echo "Platform layer: boot guard failed; refusing to serve the platform tree." >&2 exit 1 fi - # A fast-forward / rebase advanced main, so the served sha the /api/version and + # A fast-forward / merge advanced main, so the served sha the /api/version and # /api/debug/serving routes report (written to /tmp/serving-sha above) must # reflect the reconciled HEAD, not the pre-reconcile clone tip. _served_sha=$(su -s /bin/sh mobius -c \ diff --git a/backend/scripts/seed-skills/theming.md b/backend/scripts/seed-skills/theming.md index 8ece1b402..f7a5d79c4 100644 --- a/backend/scripts/seed-skills/theming.md +++ b/backend/scripts/seed-skills/theming.md @@ -89,7 +89,8 @@ git diff -- frontend/src frontend/public | head -80 touch /data/platform/frontend/src/path/to/changed-file.jsx ``` -Pick up the changes through the platform-apply flow (rebase onto `origin/main`) rather than hand-copying files — see `contributing.md`. +Pick up the changes through the platform-apply flow (merge the reviewed +`origin/main` target) rather than hand-copying files — see `contributing.md`. --- diff --git a/backend/tests/test_chats_stream_steer.py b/backend/tests/test_chats_stream_steer.py index e9fe048eb..feae3e9be 100644 --- a/backend/tests/test_chats_stream_steer.py +++ b/backend/tests/test_chats_stream_steer.py @@ -561,6 +561,198 @@ async def split_for_steer(self, rows, consume): asyncio.run(_run()) +def test_seal_publishes_the_cut_on_the_sinks_own_broadcast(): + """The cut goes to the broadcast the SINK holds, never to a fresh lookup. + + `_seal_steer_split` also runs from the turn-end `finally`, by which point a + successor turn can already have registered a NEW broadcast for the same chat. + Resolving by chat_id there would strand the cut in an event log no client is + reading: A1's blocks live in the old log, so the client would never re-base + and would paint the continuation onto the sealed segment for the rest of the + turn. Also covers a leaner writer ack (no `stored_messages`): the cut still + names the buffered rows rather than going out empty. + """ + from app.broadcast import create_broadcast, get_broadcast + from app.claude_sdk_runner import _seal_steer_split + + chat_id = "sealbroadcast" + handle = _make_active_claude_client(chat_id) + turn_bc = create_broadcast(chat_id) + + async def _run(): + handle._steer_user_msgs = [ + {"role": "user", "content": "Q2", "ts": 10, "cid": "c-q2"} + ] + handle._steer_consume_cids = ["c-q2"] + + class _SinkLike: + """Mirrors `_ChatEventSink`: holds the broadcast it was built with.""" + + def __init__(self, bc): + self.bc = bc + + async def split_for_steer(self, rows, consume): + # The writer ack shape without the echoed rows. + return {"pending": []} + + # A successor turn registers its own broadcast before this seal runs. + successor_bc = create_broadcast(chat_id) + assert get_broadcast(chat_id) is successor_bc + assert successor_bc is not turn_bc + + await _seal_steer_split(_SinkLike(turn_bc), handle, chat_id) + + assert [e.get("type") for e in successor_bc.event_log] == [] + cuts = [e for e in turn_bc.event_log if e.get("type") == "steered_into_turn"] + assert len(cuts) == 1 + assert [m["content"] for m in cuts[0]["messages"]] == ["Q2"] + assert [m["cid"] for m in cuts[0]["messages"]] == ["c-q2"] + + asyncio.run(_run()) + + +def test_writer_dedup_still_publishes_the_committed_cut(): + """An empty stored-row echo does not undo the split that just committed. + + `split_for_steer` seals A1 and resets the sink BEFORE the writer appends the + steered row. `stored_messages: []` means only that cid dedup found the row + already in the durable transcript; it does not mean the A1/A2 boundary was + skipped. Suppressing the cut here would leave the client appending A2 to the + segment the server has already sealed. The handed row still supplies the cid + that retires its tray entry and identifies the already-durable user turn. + """ + from app.broadcast import create_broadcast + from app.claude_sdk_runner import _seal_steer_split + + chat_id = "sealdedup" + handle = _make_active_claude_client(chat_id) + turn_bc = create_broadcast(chat_id) + + async def _run(): + row = {"role": "user", "content": "Q2", "ts": 10, "cid": "c-q2"} + handle._steer_user_msgs = [row] + handle._steer_consume_cids = ["c-q2"] + + class _SinkLike: + def __init__(self, bc): + self.bc = bc + + async def split_for_steer(self, rows, consume): + assert rows == [row] + assert consume == ["c-q2"] + # The durable writer already has this cid, but the sink-side split still + # sealed A1 and reset its accumulator for A2. + return {"stored_messages": []} + + await _seal_steer_split(_SinkLike(turn_bc), handle, chat_id) + + cuts = [e for e in turn_bc.event_log if e.get("type") == "steered_into_turn"] + assert len(cuts) == 1 + assert cuts[0]["messages"][0]["cid"] == "c-q2" + assert handle._steer_user_msgs == [] + assert handle._steer_consume_cids == [] + + asyncio.run(_run()) + + +def test_a_failing_publisher_cannot_escape_the_seal(): + """Announcing the cut must never raise out of `_seal_steer_split`. + + The turn-end `finally` awaits this function BEFORE it unregisters the handle + and disconnects the client, so an escaping exception would strand a live + handle in the registry and leave the chat looking permanently busy. The split + has already COMMITTED by the time the cut is published, so a broken publisher + is a notification loss, not a durability one — exactly the asymmetry the + missing-publisher branch already takes. Swallow it, log it, and still consume + the sealed rows so the turn-end retry does not double-append them. + """ + from app.claude_sdk_runner import _seal_steer_split + + chat_id = "sealpublishfail" + handle = _make_active_claude_client(chat_id) + + async def _run(): + handle._steer_user_msgs = [ + {"role": "user", "content": "Q2", "ts": 10, "cid": "c-q2"} + ] + handle._steer_consume_cids = ["c-q2"] + + class _ExplodingBroadcast: + def publish(self, event): + raise RuntimeError("broadcast is gone") + + class _SinkLike: + def __init__(self, bc): + self.bc = bc + self.splits = 0 + + async def split_for_steer(self, rows, consume): + self.splits += 1 + return {"stored_messages": list(rows)} + + sink = _SinkLike(_ExplodingBroadcast()) + await _seal_steer_split(sink, handle, chat_id) + + assert sink.splits == 1 + # The rows were committed, so they must not be re-appended by the retry. + assert handle._steer_user_msgs == [] + assert handle._steer_consume_cids == [] + + asyncio.run(_run()) + + +def test_stop_drops_the_buffered_steer_instead_of_appending_it(): + """A hard Stop abandons a deferred steer ENTIRELY. + + Stop's contract: `/chat/stop` clears `chat.pending_messages`, reports the + cleared cids, and the client re-sends exactly them as one fresh turn. A + deferred steer's row is still IN that queue (its split never ran), so if the + runner kept the row buffered, the turn-end seal appended it to the turn Stop + had just killed while the client re-sent it — the same message twice, once + interrupted and once answered. `interrupt()` therefore clears the + transcript-side buffer too, which makes the finally's seal a no-op. + """ + from app.claude_sdk_runner import ActiveClaudeClient, _seal_steer_split + + class _Client: + async def interrupt(self): + return None + + async def _run(): + # Built inside THIS loop: interrupt() waits on `_finished`, which is + # loop-bound, so a handle constructed in a throwaway loop cannot be awaited + # here. mark_finished() stands in for the runner's own teardown. + handle = ActiveClaudeClient(_Client(), chat_id="stopsteer") + handle.mark_finished() + handle.pending_steer = ["Q2"] + handle._steer_requested = True + handle._steer_user_msgs = [ + {"role": "user", "content": "Q2", "ts": 10, "cid": "c-q2"} + ] + handle._steer_consume_cids = ["c-q2"] + + await handle.interrupt() + + assert handle.pending_steer == [] + assert handle._steer_requested is False + assert handle._steer_user_msgs == [] + assert handle._steer_consume_cids == [] + + # The turn-end catch-all now has nothing to append. + class _Bc: + def __init__(self): + self.splits = 0 + + async def split_for_steer(self, rows, consume): + self.splits += 1 + + bc = _Bc() + await _seal_steer_split(bc, handle, "stopsteer") + assert bc.splits == 0 + + asyncio.run(_run()) + + def test_claude_force_steer_defers_to_runner_and_reorders(client, auth): """A Claude fast-forward (force_steer) defers its split to the runner, same as an ordinary steer, so the fast-forwarded rows land AFTER the sealed @@ -974,6 +1166,12 @@ def test_steers_into_live_claude_turn_reserves_durable_pending( assert res.status_code == 202, res.text assert res.json()["status"] == "steered" + # The split is deferred, so the response says so and echoes the row as the + # still-queued row it is; the client keeps showing it until the cut. + assert res.json()["cut_deferred"] is True + assert [m["content"] for m in res.json()["pending_messages"]] == [ + "actually use blue" + ] chat = _read_chat(chat_id) assert [m["content"] for m in chat.pending_messages] == ["actually use blue"] @@ -986,21 +1184,14 @@ def test_steers_into_live_claude_turn_reserves_durable_pending( assert cid_of(handle._steer_user_msgs[0]) == reserved_cid assert handle._steer_consume_cids == [reserved_cid] - # A `steered_into_turn` event was broadcast for the inline render. + # NO event at HTTP arrival on the deferred path. The 202's own + # `pending_messages` (asserted above) is the single signal that keeps the row + # visible; the CUT (`steered_into_turn`) belongs to the runner's seal — see + # test_claude_steer_cut_event_is_published_at_the_seal_not_at_http_arrival. + # A second "accepted" event reconciling the same tray would be a parallel + # channel racing the response it duplicates. bc = get_broadcast(chat_id) - steered_events = [ - e for e in bc.event_log if e.get("type") == "steered_into_turn" - ] - assert len(steered_events) == 1 - assert steered_events[0]["content"] == "actually use blue" - assert steered_events[0]["messages"] == [ - { - "role": "user", - "ts": handle._steer_user_msgs[0]["ts"], - "cid": reserved_cid, - "content": "actually use blue", - } - ] + assert bc.event_log == [] def test_claude_runner_splits_steer_at_boundary_not_http_arrival( @@ -1074,6 +1265,160 @@ async def _drive_runner(): assert _read_chat(chat_id).pending_messages in (None, []) +def test_claude_steer_cut_event_is_published_at_the_seal_not_at_http_arrival( + client, auth, +): + """`steered_into_turn` is the client's only "cut the live stream here" signal, + so on the deferred (Claude) path it must be published by the runner at the + seal — AFTER every block that belongs to A1 — and never by the route. + + The regression this pins: the route published the cut at HTTP arrival while + the split stayed at the runner's interrupt boundary seconds later. Everything + Claude streamed in the gap was accumulated into the sealed A1 AND kept at the + head of the client's freshly re-based stream, so it painted twice for the rest + of the turn. The window is never empty — the AssistantMessage that triggers + the boundary interrupt is dispatched to the broadcast before the interrupt + check runs — so this duplicated on EVERY Claude steer. + """ + from app.chat import _ChatEventSink, register_active_sink + from app.claude_sdk_runner import _seal_steer_split + + chat_id = "claudecutorder" + db = SessionLocal() + try: + db.add(models.Chat( + id=chat_id, title="Claude chat", provider="claude", + messages=[{"role": "user", "content": "Q1", "ts": 1}], + agent_settings_json={"steer_enabled": True}, + )) + db.commit() + finally: + db.close() + handle = _make_active_claude_client(chat_id) + registry.register(handle) + bc = create_broadcast(chat_id) + sink = _ChatEventSink(bc, chat_id, run_token="run-cut-order") + register_active_sink(chat_id, sink) + + # A1's first block is already on the wire when the steer POST arrives. + sink.publish({"type": "text", "content": "A1 first"}) + + res = client.post( + f"/api/chats/{chat_id}/messages", + json={"content": "Q2"}, headers=auth, + ) + assert res.status_code == 202, res.text + assert res.json()["status"] == "steered" + + # HTTP arrival publishes NOTHING on the deferred path: the response body is + # the display signal. Publishing the cut here re-based the client's stream + # too early. + assert [e.get("type") for e in bc.event_log] == ["text"] + + async def _drive_runner(): + # The rest of A1 streams AFTER arrival — the duplication window. + sink.publish({"type": "text", "content": " A1 rest"}) + await _seal_steer_split(sink, handle, chat_id) + # A2's first block follows the seal. It exists here so the cut's position + # is pinned from BOTH sides: a cut that slipped in front of a continuation + # block would fold A2's head into the sealed A1 and re-base after it — + # the mirror image of the fixed bug, and invisible to a lower bound alone. + sink.publish({"type": "text", "content": "A2 head"}) + + asyncio.run(_drive_runner()) + + # The cut is published exactly once, at the seal: after every A1 block and + # before every A2 block. + cut_positions = [ + i for i, e in enumerate(bc.event_log) + if e.get("type") == "steered_into_turn" + ] + assert len(cut_positions) == 1 + # The sink coalesces contiguous text into one event per segment, so the whole + # of A1 is one event and A2's head is another. Both bounds matter: after the + # last A1 event (the fixed bug) AND before the first A2 event (its mirror + # image, which would fold A2's head into the sealed A1). + text_positions = [ + i for i, e in enumerate(bc.event_log) if e.get("type") == "text" + ] + assert [bc.event_log[i].get("content") for i in text_positions] == [ + "A1 first A1 rest", "A2 head", + ] + assert text_positions[0] < cut_positions[0] < text_positions[1] + # The cut names the DURABLE rows the split committed, so the client inserts + # the same identity the transcript now holds. + cut = bc.event_log[cut_positions[0]] + steered_row = [m for m in _read_chat(chat_id).messages if m["role"] == "user"][-1] + assert cut["messages"] == [{ + "role": "user", + "ts": steered_row["ts"], + "cid": cid_of(steered_row), + "content": "Q2", + }] + # And A1 really was sealed at the boundary the cut names. + assert [(m["role"], m.get("content")) for m in _read_chat(chat_id).messages] == [ + ("user", "Q1"), + ("assistant", "A1 first A1 rest"), + ("user", "Q2"), + ] + + +def test_codex_steer_still_publishes_the_cut_at_the_route( + client, auth, monkeypatch, +): + """Codex is unaffected by moving the Claude cut. + + Its `turn.steer()` injects into the SAME running turn, so the route's own + `_split_steer_at_route` IS the seal: signal and cut are the same instant + there. The route must therefore keep publishing `steered_into_turn` at + arrival, with the response shape unchanged (no `cut_deferred`). + """ + chat_id = "codexcutroute" + db = SessionLocal() + try: + db.add(models.Chat( + id=chat_id, title="Codex chat", provider="codex", + messages=[{"role": "user", "content": "Q1", "ts": 1}], + agent_settings_json={"steer_enabled": True}, + )) + db.commit() + finally: + db.close() + registry.register(_make_active_codex_turn(chat_id)) + sink = _register_sink_with_partial(chat_id, "run-codex-cut", "A1") + + async def _fake_steer(cid, message): + return True + + monkeypatch.setattr( + "app.codex_sdk_runner.steer_into_active_turn", _fake_steer, + ) + + res = client.post( + f"/api/chats/{chat_id}/messages", + json={"content": "Q2"}, headers=auth, + ) + assert res.status_code == 202, res.text + body = res.json() + assert body["status"] == "steered" + assert "cut_deferred" not in body + # The row left pending at the route, so the echoed queue no longer holds it. + assert body["pending_messages"] == [] + + bc = get_broadcast(chat_id) + assert [e.get("type") for e in bc.event_log] == ["steered_into_turn"] + cut = [e for e in bc.event_log if e.get("type") == "steered_into_turn"][0] + assert [m["content"] for m in cut["messages"]] == ["Q2"] + # The route sealed A1 and appended Q2 before publishing — the cut is + # truthful at the instant it is sent. + assert [(m["role"], m.get("content")) for m in _read_chat(chat_id).messages] == [ + ("user", "Q1"), + ("assistant", "A1"), + ("user", "Q2"), + ] + assert sink.assistant_blocks == [] + + def test_claude_reserved_row_survives_process_loss_and_sweep( client, auth, monkeypatch, ): diff --git a/backend/tests/test_claude_sdk_runner.py b/backend/tests/test_claude_sdk_runner.py index 7b7fcae34..acdd19d28 100644 --- a/backend/tests/test_claude_sdk_runner.py +++ b/backend/tests/test_claude_sdk_runner.py @@ -1074,7 +1074,10 @@ def test_dispatch_client_web_search_tool_result_emits_sources(): {"type": "tool_start", "tool": "WebSearch", "input": "", "tool_use_id": "t1"}, {"type": "tool_input", "tool": "WebSearch", "input": "mobius docs", "tool_use_id": "t1"}, - {"type": "tool_output", "content": result_text, "tool_use_id": "t1"}, + { + "type": "tool_output", "content": result_text, "tool_use_id": "t1", + "output_complete": True, + }, # tool_use_id binds these sources to the search that produced them, so a # batch of parallel WebSearch calls does not collapse onto one block. {"type": "tool_sources", "tool_use_id": "t1", "sources": [ diff --git a/backend/tests/test_codex_sdk_contract.py b/backend/tests/test_codex_sdk_contract.py index 402ad9caf..218e530a3 100644 --- a/backend/tests/test_codex_sdk_contract.py +++ b/backend/tests/test_codex_sdk_contract.py @@ -239,3 +239,21 @@ def test_reasoning_effort_enum_tolerates_unknown_efforts(): for value in ("high", "xhigh", "max", "ultra", "some-future-effort"): assert ReasoningEffort(value).value == value + + +def test_transport_closed_error_is_still_exposed_by_the_sdk(): + """_is_transport_death decides whether an error reaches the owner at all. + + It binds to this class by isinstance, and a stop that SIGTERMs the turn's + process group is reclassified as a clean interrupt only when the raised + exception is one. If a future SDK renames or drops the symbol, the runner + would stop recognizing its own teardown and go back to publishing a raw + provider error over the pause note — so fail here first. + """ + pytest.importorskip("openai_codex") + from openai_codex.errors import CodexError, TransportClosedError + + assert issubclass(TransportClosedError, CodexError) + # Not a RuntimeError: the runner relies on that to keep provider + # ErrorNotifications (reraised as plain RuntimeErrors) out of this branch. + assert not issubclass(TransportClosedError, RuntimeError) diff --git a/backend/tests/test_codex_sdk_runner.py b/backend/tests/test_codex_sdk_runner.py index 0404459c4..f5d16dff0 100644 --- a/backend/tests/test_codex_sdk_runner.py +++ b/backend/tests/test_codex_sdk_runner.py @@ -17,6 +17,18 @@ # - CodexRpcError: /usr/local/lib/python3.12/site-packages/openai_codex/errors.py:24 # - InvalidParamsError: /usr/local/lib/python3.12/site-packages/openai_codex/errors.py:40 +try: + from openai_codex.errors import ( + TransportClosedError as _SdkTransportClosedError, + ) +except ImportError: # pragma: no cover - SDK is an optional install + # openai-codex is installed in its own Docker step, not via + # requirements.txt, so this module must still import without it. The real + # symbol's continued existence is pinned by test_codex_sdk_contract, which + # is where its disappearance should be reported. + class _SdkTransportClosedError(Exception): + pass + class _FakeBroadcast: def __init__(self): @@ -177,6 +189,7 @@ def __init__(self, code: int, message: str, data=None): "ReasoningSummaryTextDeltaNotification": _Dummy, "ReasoningTextDeltaNotification": _Dummy, "ThreadTokenUsageUpdatedNotification": _Dummy, + "TransportClosedError": _SdkTransportClosedError, "TurnCompletedNotification": _FakeTurnCompletedNotification, "TurnStatus": _FakeTurnStatus, "WebSearchThreadItem": _Dummy, @@ -205,8 +218,9 @@ def test_stamp_tool_use_id_uses_stable_item_id(): def test_tool_completed_events_emit_output_before_end(): class CommandExecutionThreadItem: - def __init__(self, output: str): + def __init__(self, output: str, exit_code: int = 0): self.aggregated_output = output + self.exit_code = exit_code sdk = {"CommandExecutionThreadItem": CommandExecutionThreadItem} sdk.update({ @@ -222,7 +236,20 @@ def __init__(self, output: str): ) assert events == [ - {"type": "tool_output", "content": "hello"}, + { + "type": "tool_output", "content": "hello", + "output_complete": True, "output_exit_code": 0, + }, + {"type": "tool_end"}, + ] + + assert codex_sdk_runner._tool_completed_events( + CommandExecutionThreadItem("", exit_code=7), sdk, + ) == [ + { + "type": "tool_output", "content": "", + "output_complete": True, "output_exit_code": 7, + }, {"type": "tool_end"}, ] @@ -568,6 +595,50 @@ def test_is_closed_turn_error_does_not_treat_arbitrary_oserror_as_closed(): assert codex_sdk_runner._is_closed_turn_error(OSError("disk full")) is False +def test_is_transport_death_rejects_provider_runtime_error_about_not_running( + monkeypatch, +): + """The narrow predicate is the whole reason the two exist separately. + + The runner reraises every non-retryable provider ErrorNotification as a + plain `RuntimeError(message)`, and those messages routinely say "is not + running". _is_closed_turn_error accepts that text because the only cost + of a false positive there is a refused steer; the except-path + reclassification must not, or a stop racing a genuine MCP failure would + swallow the only report the owner ever gets. Widening the guard back to + _is_closed_turn_error fails here. + """ + sdk = _fake_sdk(async_codex_cls=object) + monkeypatch.setattr(codex_sdk_runner, "_sdk_imports", lambda: sdk) + + provider_fault = RuntimeError("MCP server 'x' is not running") + + assert codex_sdk_runner._is_transport_death(provider_fault) is False + assert codex_sdk_runner._is_closed_turn_error(provider_fault) is True + + +def test_is_transport_death_binds_to_the_sdk_class_not_to_its_name(): + """Uses the installed SDK, because the binding is what is under test. + + isinstance against the symbol `_sdk_imports()` exposes means a genuine + subclass matches and a same-named impostor does not — neither of which a + class-name comparison could get right. + """ + pytest.importorskip("openai_codex") + + class _MoreSpecificTransportError(_SdkTransportClosedError): + pass + + impostor = type("TransportClosedError", (Exception,), {}) + + assert codex_sdk_runner._is_transport_death( + _MoreSpecificTransportError("closed stdout") + ) is True + assert codex_sdk_runner._is_transport_death( + impostor("closed stdout") + ) is False + + def test_run_codex_sdk_turn_resume_mismatch_returns_error(monkeypatch): mismatched_thread = _FakeThread("actual-thread", _FakeTurnHandle()) @@ -1288,6 +1359,301 @@ def _mark_finished(self): assert mark_finished_calls == [True] +class _KilledTransportError(_SdkTransportClosedError): + """A subclass of the SDK's own TransportClosedError. + + Subclassing the real symbol rather than redeclaring its name is the point: + the runner reclassifies on isinstance, so a look-alike would prove nothing + and a genuine subclass would not match a name check. It is also not a + RuntimeError, which keeps the narrow transport branch distinguishable from + the looser bare-RuntimeError-text branch that belongs to + _is_closed_turn_error alone. + """ + + +def _run_turn_whose_stream_dies( + monkeypatch, + exc: Exception, + *, + on_register=None, + should_abort=None, + notifications=None, + sdk_patch=None, +): + """Runs one turn whose stream raises `exc`, optionally mid-teardown. + + `on_register` fires against the handle the runner has just registered — + the same window in which the real Stop / stall watchdog reaches a live + turn — so a test can mark the teardown as ours before the stream dies. + `notifications` are delivered before the death, so a test can give the + turn something to have spent; `sdk_patch` supplies the payload classes + those notifications need to be recognized as. + """ + turn_handle = _FakeTurnHandle(notifications, stream_exc=exc) + thread = _FakeThread("thread-1", turn_handle) + + class FakeAsyncCodex: + def __init__(self, config=None): + self.config = config + + async def __aenter__(self): + return self + + async def __aexit__(self, _exc_type, _exc, _tb): + return None + + async def thread_start(self, *_args, **_kwargs): + return thread + + sdk = _fake_sdk(FakeAsyncCodex) + sdk.update(sdk_patch or {}) + monkeypatch.setattr(codex_sdk_runner, "_sdk_imports", lambda: sdk) + + if on_register is not None: + original_register = registry.register + + def _register(handle): + on_register(handle) + return original_register(handle) + + monkeypatch.setattr(registry, "register", _register) + + bc = _FakeBroadcast() + result = asyncio.run( + codex_sdk_runner.run_codex_sdk_turn( + user_message="hello", + session_id=None, + base_env={}, + cwd="/tmp", + chat_id="chat-1", + bc=bc, + pending_questions={}, + db=None, + **({"should_abort": should_abort} if should_abort else {}), + ) + ) + return result, bc + + +def _mark_interrupted(handle): + handle._interrupt_requested = True + + +def test_run_codex_sdk_turn_reports_self_requested_kill_as_interrupted( + monkeypatch, caplog, +): + """A stop we asked for must not read as a provider failure. + + Stop and the stall watchdog both interrupt first and then, on timeout, + SIGTERM the turn's private process group — so the transport dies + mid-stream instead of delivering turn/completed. Reporting the resulting + "closed stdout" as an error is wrong twice over: it blames Codex for our + own teardown, and the error block overwrites the stop/stall note in the + transcript and strips its one-tap Resume. + """ + result, bc = _run_turn_whose_stream_dies( + monkeypatch, + _KilledTransportError("Codex process closed stdout. stderr_tail="), + on_register=_mark_interrupted, + ) + + assert result["error"] is None + assert result["terminal_status"] == "interrupted" + assert [e for e in bc.events if e.get("type") == "error"] == [] + assert registry.get_handle("chat-1", RunnerKind.CODEX_SDK) is None + assert any( + record.levelname == "WARNING" + and "Codex transport closed by our own stop" in record.message + and "closed stdout" in record.message + for record in caplog.records + ) + + +def test_run_codex_sdk_turn_self_requested_kill_still_reports_usage( + monkeypatch, +): + """A stop we asked for is a clean interrupt, but the tokens still count. + + Usage delivered before the self-kill must survive the transport-death + reclassification: the interrupted (error=None) exit routes through + with_usage just like every other exit, so a stopped turn is not free in + the budget ledger. + """ + class TokenUsageUpdated: + def __init__(self, token_usage): + self.token_usage = token_usage + + class Usage: + def __init__(self, total_tokens): + self.last = SimpleNamespace( + input_tokens=200, cached_input_tokens=100, output_tokens=100, + reasoning_output_tokens=50, total_tokens=300, + ) + self.total = SimpleNamespace( + input_tokens=1_000, cached_input_tokens=400, output_tokens=100, + reasoning_output_tokens=50, total_tokens=total_tokens, + ) + self.model_context_window = 200_000 + + def model_dump(self, **_kwargs): + return { + "last": vars(self.last), + "total": vars(self.total), + "modelContextWindow": self.model_context_window, + } + + result, _bc = _run_turn_whose_stream_dies( + monkeypatch, + _KilledTransportError("Codex process closed stdout. stderr_tail="), + on_register=_mark_interrupted, + notifications=[ + SimpleNamespace( + method="thread/tokenUsage/updated", + payload=TokenUsageUpdated(Usage(1_100)), + ), + ], + sdk_patch={"ThreadTokenUsageUpdatedNotification": TokenUsageUpdated}, + ) + + assert result["error"] is None + assert result["terminal_status"] == "interrupted" + assert result["usage"]["total"]["total_tokens"] == 1_100 + assert "usage_metrics" in result + + +def test_run_codex_sdk_turn_unrequested_transport_death_stays_an_error( + monkeypatch, +): + """The same transport death with no stop pending is still a real failure. + + Only a teardown we asked for may be reclassified, so a provider-side + crash keeps its error. + """ + result, _bc = _run_turn_whose_stream_dies( + monkeypatch, + _KilledTransportError("Codex process closed stdout. stderr_tail="), + ) + + assert result["error"] == "Codex process closed stdout. stderr_tail=" + assert result.get("terminal_status") is None + assert registry.get_handle("chat-1", RunnerKind.CODEX_SDK) is None + + +def test_run_codex_sdk_turn_non_transport_failure_during_stop_stays_an_error( + monkeypatch, +): + """A pending stop must not launder an unrelated bug into a clean stop. + + Without this, any defect that happens to fire inside a stop window + returns "interrupted" with no error and no log line — invisible in both + the transcript and chat.log. + """ + result, _bc = _run_turn_whose_stream_dies( + monkeypatch, + ValueError("unexpected notification payload"), + on_register=_mark_interrupted, + ) + + assert result["error"] == "unexpected notification payload" + assert result.get("terminal_status") is None + + +def test_run_codex_sdk_turn_provider_fault_during_stop_keeps_its_error( + monkeypatch, +): + """A provider fault that merely looks closed-ish must still be reported. + + Non-retryable provider errors reach this except path as plain + RuntimeErrors carrying the app-server's own words, and "is not running" + is ordinary phrasing for a broken MCP server. Only the transport itself + dying may be reclassified: if the guard is widened to + _is_closed_turn_error, a stop in flight turns a real failure into a + silent clean interrupt. + """ + result, _bc = _run_turn_whose_stream_dies( + monkeypatch, + RuntimeError("MCP server 'x' is not running"), + on_register=_mark_interrupted, + ) + + assert result["error"] == "MCP server 'x' is not running" + assert result.get("terminal_status") is None + + +def test_run_codex_sdk_turn_failed_turn_still_reports_what_it_spent( + monkeypatch, +): + """Tokens burned before a crash are still tokens burned. + + Every other exit routes its usage through with_usage; the raw error exit + used to drop it, so a turn that streamed for minutes and then died looked + free in the budget ledger. + """ + class TokenUsageUpdated: + def __init__(self, token_usage): + self.token_usage = token_usage + + class Usage: + def __init__(self, total_tokens): + self.last = SimpleNamespace( + input_tokens=200, cached_input_tokens=100, output_tokens=100, + reasoning_output_tokens=50, total_tokens=300, + ) + self.total = SimpleNamespace( + input_tokens=1_000, cached_input_tokens=400, output_tokens=100, + reasoning_output_tokens=50, total_tokens=total_tokens, + ) + self.model_context_window = 200_000 + + def model_dump(self, **_kwargs): + return { + "last": vars(self.last), + "total": vars(self.total), + "modelContextWindow": self.model_context_window, + } + + result, _bc = _run_turn_whose_stream_dies( + monkeypatch, + ValueError("unexpected notification payload"), + notifications=[ + SimpleNamespace( + method="thread/tokenUsage/updated", + payload=TokenUsageUpdated(Usage(1_100)), + ), + ], + sdk_patch={"ThreadTokenUsageUpdatedNotification": TokenUsageUpdated}, + ) + + assert result["error"] == "unexpected notification payload" + assert result["usage"]["total"]["total_tokens"] == 1_100 + # One update means no thread delta to compute, so the exact numbers are + # normalize_codex_usage's business; what this test owns is that the error + # exit carries the metrics at all. + assert "usage_metrics" in result + + +def test_run_codex_sdk_turn_superseded_generation_kill_is_interrupted( + monkeypatch, +): + """A newer turn taking over the chat is also Möbius ending this one. + + This leg needs no ActiveCodexTurn at all, so it is the one that reaches + a teardown during startup — pinned here so a later narrowing of + stop_requested() to the interrupt flag alone goes red. + """ + superseded = {"value": False} + + result, _bc = _run_turn_whose_stream_dies( + monkeypatch, + _KilledTransportError("Codex process closed stdout. stderr_tail="), + on_register=lambda _handle: superseded.update(value=True), + should_abort=lambda: superseded["value"], + ) + + assert result["error"] is None + assert result["terminal_status"] == "interrupted" + + def test_run_codex_sdk_turn_error_notification_will_retry_continues(monkeypatch): class AgentMessageDeltaNotification: def __init__(self, delta: str): diff --git a/backend/tests/test_memory_recall.py b/backend/tests/test_memory_recall.py new file mode 100644 index 000000000..18444ac5d --- /dev/null +++ b/backend/tests/test_memory_recall.py @@ -0,0 +1,371 @@ +"""Memory recall citations: identification, parsing, and survival on read. + +The behaviour under test is what lets an owner tell three states apart — +the turn recalled these notes / it looked and Memory had nothing / it never +looked. The third is the absence of a citation, so the tests below care as +much about what is NOT stamped as about what is. +""" + +import json + +from app.chat_transcript import ( + _compact_activity_item, + _compact_activity_run, + _distinctive_activity, +) +from app.chat import _ChatEventSink +from app.events import process_event +from app.memory_recall import ( + MAX_RECALL_NOTES, + RECALL_EMPTY, + RECALL_FAILED, + RECALL_HIT, + RECALL_SEARCHING, + recall_from_command, + recall_from_result, +) + +MEMORY_CMD = 'python3 /data/apps/memory/memory_search.py "what does he prefer" "chat-1"' + +# Synthetic notes. Fixtures here become a public diff, so they must never carry +# anything from a real owner's graph — a memory note is personal by definition. +HIT_OUTPUT = """Relevant memories: +- Apps render in a sandboxed frame: Each mini-app runs isolated. [notes/apps-render-in-a-sandboxed-frame.md] +- Theme variables are shared: Colors come from one stylesheet. [notes/theme-variables-are-shared.md] +FILES: notes/apps-render-in-a-sandboxed-frame.md, notes/theme-variables-are-shared.md +MOBIUS_MEMORY_RESULT_V1:{"status":"hit","notes":[{"id":"apps-render-in-a-sandboxed-frame","path":"notes/apps-render-in-a-sandboxed-frame.md","title":"Apps render in a sandboxed frame","excerpt":"Each mini-app runs isolated."},{"id":"theme-variables-are-shared","path":"notes/theme-variables-are-shared.md","title":"Theme variables are shared","excerpt":"Colors come from one stylesheet."}]}""" +EMPTY_OUTPUT = """No relevant memories. +MOBIUS_MEMORY_RESULT_V1:{"status":"empty"}""" +FAILED_OUTPUT = """Memory lookup failed. +MOBIUS_MEMORY_RESULT_V1:{"status":"failed"}""" + + +# --- identification ------------------------------------------------------- + +def test_a_memory_search_command_is_identified_as_a_lookup(): + assert recall_from_command(MEMORY_CMD) == { + "status": RECALL_SEARCHING, "app_slug": "memory", + } + + +def test_a_command_merely_mentioning_memory_search_is_not_a_lookup(): + # Identification gates everything downstream, so a false positive here would + # mint citations from an unrelated command's output. + # Every one of these is an ordinary thing to do WHILE working on Memory, and + # each names the script without running it. + assert recall_from_command("grep -rn memory_search.py /data/platform") is None + assert recall_from_command("cat /data/apps/memory/memory_search.py") is None + assert recall_from_command("wc -l memory_search.py") is None + assert recall_from_command("ls -la /data/apps/memory/memory_search.py") is None + assert recall_from_command("vim memory_search.py") is None + assert recall_from_command("python3 -m py_compile app/memory_search.py") is None + assert recall_from_command("echo memory_search.python") is None + assert recall_from_command("ls /data/apps/memory/") is None + assert recall_from_command("") is None + assert recall_from_command(None) is None + + +def test_the_documented_simple_invocation_is_recognized(): + assert recall_from_command(MEMORY_CMD) is not None + assert recall_from_command( + 'MEMORY_READER_PROVIDER=none python3 -u ' + '/data/apps/memory-2/memory_search.py "q" "chat-1"' + ) == {"status": RECALL_SEARCHING, "app_slug": "memory-2"} + + +def test_shell_composition_and_non_memory_paths_are_rejected_conservatively(): + assert recall_from_command( + 'python3 /data/apps/memory/memory_search.py "q"' + ) is None + assert recall_from_command( + 'cd /x && python3 /data/apps/memory/memory_search.py "q"' + ) is None + assert recall_from_command('python3 ./memory_search.py "q"') is None + assert recall_from_command('python3 /a/b/memory_search.py "q"') is None + assert recall_from_command( + 'python3 /data/apps/memory/memory_search.py "q" > /tmp/result' + ) is None + + +def test_trailing_arguments_and_newline_commands_cannot_mint_recall_metadata(): + assert recall_from_command(MEMORY_CMD + ' "unexpected"') is None + assert recall_from_command( + MEMORY_CMD + '\nprintf \'MOBIUS_MEMORY_RESULT_V1:{"status":"hit"}\\n\'' + ) is None + + +# --- parsing -------------------------------------------------------------- + +def test_a_successful_lookup_cites_the_notes_it_opened(): + recall = recall_from_result(HIT_OUTPUT, 0) + assert recall["status"] == RECALL_HIT + assert [note["id"] for note in recall["notes"]] == [ + "apps-render-in-a-sandboxed-frame", "theme-variables-are-shared", + ] + assert recall["notes"][0]["title"] == "Apps render in a sandboxed frame" + assert recall["notes"][0]["excerpt"] == "Each mini-app runs isolated." + + +def test_a_citation_keeps_the_graph_node_id_when_it_differs_from_the_file(): + recall = recall_from_result( + 'MOBIUS_MEMORY_RESULT_V1:{"status":"hit","notes":[' + '{"id":"canonical-node","path":"notes/readable-filename.md",' + '"title":"Canonical node"}]}', + 0, + ) + + assert recall["notes"] == [{ + "id": "canonical-node", + "path": "notes/readable-filename.md", + "title": "Canonical node", + }] + + +def test_an_unsafe_or_missing_graph_node_id_falls_back_to_the_file_stem(): + recall = recall_from_result( + 'MOBIUS_MEMORY_RESULT_V1:{"status":"hit","notes":[' + '{"id":"../escape","path":"notes/safe-fallback.md"},' + '{"path":"notes/legacy-note.md"}]}', + 0, + ) + + assert [note["id"] for note in recall["notes"]] == [ + "safe-fallback", "legacy-note", + ] + + +def test_a_lookup_that_found_nothing_says_so(): + assert recall_from_result(EMPTY_OUTPUT, 0) == {"status": RECALL_EMPTY} + + +def test_a_carved_output_keeps_the_structured_tail_result(): + carved = "presentation head\n…[large middle carved]…\n" + HIT_OUTPUT.splitlines()[-1] + recall = recall_from_result(carved, 0) + assert [note["id"] for note in recall["notes"]] == [ + "apps-render-in-a-sandboxed-frame", "theme-variables-are-shared", + ] + + +def test_unreadable_or_failed_results_are_explicit_failures(): + for body in ("", " ", "some unrelated text", "Traceback (most recent call last):"): + assert recall_from_result(body, 0) == {"status": RECALL_FAILED} + assert recall_from_result(FAILED_OUTPUT, 1) == {"status": RECALL_FAILED} + assert recall_from_result(HIT_OUTPUT, 1) == {"status": RECALL_FAILED} + + +def test_a_citation_path_may_not_escape_the_graph(): + recall = recall_from_result( + 'MOBIUS_MEMORY_RESULT_V1:{"status":"hit","notes":[' + '{"path":"../../etc/passwd"},{"path":"/abs/x.md"},' + '{"path":"notes/../secret.md"},{"path":"notes/ok.md"}]}', + 0, + ) + assert [note["path"] for note in recall["notes"]] == ["notes/ok.md"] + + +def test_repeated_and_excessive_citations_are_bounded(): + notes = [{"path": "notes/dup.md"}, {"path": "notes/dup.md"}] + [ + {"path": f"notes/n{i}.md"} for i in range(40) + ] + recall = recall_from_result( + "MOBIUS_MEMORY_RESULT_V1:" + json.dumps({"status": "hit", "notes": notes}), + 0, + ) + assert len(recall["notes"]) == MAX_RECALL_NOTES + assert recall["notes"][0]["path"] == "notes/dup.md" + assert len({note["path"] for note in recall["notes"]}) == len(recall["notes"]) + + +def test_the_last_structured_result_line_wins(): + recall = recall_from_result( + 'MOBIUS_MEMORY_RESULT_V1:{"status":"hit","notes":[{"path":"notes/stale.md"}]}\n' + 'MOBIUS_MEMORY_RESULT_V1:{"status":"hit","notes":[{"path":"notes/real.md"}]}', + 0, + ) + assert [note["path"] for note in recall["notes"]] == ["notes/real.md"] + + +# --- the block carries it through persistence ------------------------------ + +def _tool_blocks(recall_in, recall_out, output=HIT_OUTPUT, recall_on_start=False): + blocks: list = [] + start = {"type": "tool_start", "tool": "Bash", "input": MEMORY_CMD, + "tool_use_id": "t1"} + if recall_on_start and recall_in is not None: + start["recall"] = recall_in + process_event(start, blocks) + event_in = {"type": "tool_input", "tool_use_id": "t1", "input": MEMORY_CMD} + if not recall_on_start and recall_in is not None: + event_in["recall"] = recall_in + process_event(event_in, blocks) + event_out = {"type": "tool_output", "tool_use_id": "t1", "content": output} + if recall_out is not None: + event_out["recall"] = recall_out + process_event(event_out, blocks) + return blocks + + +def test_the_lookup_marker_reaches_the_persisted_block_and_then_settles(): + blocks = _tool_blocks( + {"status": RECALL_SEARCHING}, + {"status": RECALL_HIT, "notes": [{"id": "a", "path": "notes/a.md", "title": "A"}]}, + ) + assert blocks[0]["recall"]["status"] == RECALL_HIT + assert blocks[0]["recall"]["notes"][0]["id"] == "a" + + +def test_codex_tool_start_carries_the_lookup_marker_without_tool_input(): + blocks = _tool_blocks( + {"status": RECALL_SEARCHING}, + {"status": RECALL_HIT, "notes": [{"id": "a", "path": "notes/a.md"}]}, + recall_on_start=True, + ) + assert blocks[0]["recall"]["status"] == RECALL_HIT + + +def test_the_claude_path_does_not_double_stamp_a_single_lookup(): + # Simulate a runner that supplies the memory_search command on BOTH the + # tool_start and a following tool_input for the same tool_use_id. The block + # must be stamped once: the second phase sees the block already carries a + # recall marker and is skipped, so no duplicate/overwriting stamp occurs. + sink = object.__new__(_ChatEventSink) + sink.assistant_blocks = [] + start = {"type": "tool_start", "tool": "Bash", "input": MEMORY_CMD, + "tool_use_id": "t1"} + sink._stamp_memory_recall(start) + process_event(start, sink.assistant_blocks) + assert sink.assistant_blocks[0]["recall"]["status"] == RECALL_SEARCHING + follow = {"type": "tool_input", "tool_use_id": "t1", "input": MEMORY_CMD} + sink._stamp_memory_recall(follow) + assert "recall" not in follow + process_event(follow, sink.assistant_blocks) + assert sink.assistant_blocks[0]["recall"]["status"] == RECALL_SEARCHING + + +def test_partial_output_does_not_settle_the_lookup_before_completion(): + blocks: list = [] + process_event({ + "type": "tool_start", "tool": "Bash", "input": MEMORY_CMD, + "tool_use_id": "t1", "recall": {"status": RECALL_SEARCHING}, + }, blocks) + process_event({"type": "tool_output", "tool_use_id": "t1", "content": "partial"}, blocks) + assert blocks[0]["recall"]["status"] == RECALL_SEARCHING + process_event({ + "type": "tool_output", "tool_use_id": "t1", "content": HIT_OUTPUT, + "recall": {"status": RECALL_HIT, "notes": [{"id": "a", "path": "notes/a.md"}]}, + }, blocks) + assert blocks[0]["recall"]["status"] == RECALL_HIT + + +def _sink_lifecycle(events): + sink = object.__new__(_ChatEventSink) + sink.assistant_blocks = [] + for event in events: + sink._stamp_memory_recall(event) + process_event(event, sink.assistant_blocks) + return sink.assistant_blocks[0]["recall"] + + +def test_claude_and_codex_lifecycles_settle_to_identical_recall_metadata(): + final = { + "type": "tool_output", "tool_use_id": "t1", "content": HIT_OUTPUT, + "output_complete": True, "output_exit_code": 0, + } + codex = _sink_lifecycle([ + {"type": "tool_start", "tool": "Bash", "input": MEMORY_CMD, + "tool_use_id": "t1"}, + {"type": "tool_output", "tool_use_id": "t1", "content": "partial"}, + dict(final), + ]) + claude = _sink_lifecycle([ + {"type": "tool_start", "tool": "Bash", "input": "", + "tool_use_id": "t1"}, + {"type": "tool_input", "input": MEMORY_CMD, "tool_use_id": "t1"}, + dict(final), + ]) + assert codex == claude + assert codex["status"] == RECALL_HIT + assert [note["id"] for note in codex["notes"]] == [ + "apps-render-in-a-sandboxed-frame", "theme-variables-are-shared", + ] + assert {note["app_slug"] for note in codex["notes"]} == {"memory"} + + +def test_an_ordinary_command_gains_no_recall_field(): + blocks = _tool_blocks(None, None, output="total 0\n") + assert "recall" not in blocks[0] + + +# --- survival through the read-side projection ----------------------------- + +def test_consulting_memory_is_its_own_activity_beat(): + assert _distinctive_activity({"type": "tool", "tool": "Bash", + "recall": {"status": RECALL_HIT, "notes": []}}) + assert not _distinctive_activity({"type": "tool", "tool": "Bash"}) + + +def test_a_failed_lookup_remains_an_activity_without_citations(): + assert _distinctive_activity({ + "type": "tool", "tool": "Bash", "recall": {"status": RECALL_FAILED}, + }) + + +def test_the_compacted_line_still_knows_what_it_recalled(): + # Without this the beat renders live and reverts to "Ran a command" on the + # next chat load, which is worse for trust than never having shown it. + item = _compact_activity_item({ + "type": "tool", "tool": "Bash", "status": "done", + "input": MEMORY_CMD, + "recall": {"status": RECALL_HIT, "notes": [{"id": "a", "path": "notes/a.md"}]}, + }) + assert item["recall"]["notes"][0]["id"] == "a" + + +def test_citations_roll_up_so_a_folded_run_keeps_them(): + # _compact_activity_entries keeps only two entries per tool name, so a third + # lookup's notes exist ONLY on the run summary. + blocks = [ + (i, {"type": "tool", "tool": "Bash", "status": "done", + "recall": {"status": RECALL_HIT, + "notes": [{"id": f"n{i}", "path": f"notes/n{i}.md"}]}}) + for i in range(3) + ] + run = _compact_activity_run(blocks, message_index=0) + assert [note["id"] for note in run["recall"]["notes"]] == ["n0", "n1", "n2"] + assert run["recall"]["status"] == RECALL_HIT + + +def test_a_run_with_no_lookup_carries_no_recall_key(): + blocks = [(0, {"type": "tool", "tool": "Bash", "status": "done"})] + assert "recall" not in _compact_activity_run(blocks, message_index=0) + + +def test_a_remembered_note_outranks_an_empty_probe_in_the_same_run(): + blocks = [ + (0, {"type": "tool", "tool": "Bash", "status": "done", + "recall": {"status": RECALL_EMPTY}}), + (1, {"type": "tool", "tool": "Bash", "status": "done", + "recall": {"status": RECALL_HIT, + "notes": [{"id": "a", "path": "notes/a.md"}]}}), + ] + run = _compact_activity_run(blocks, message_index=0) + assert run["recall"]["status"] == RECALL_HIT + assert [note["id"] for note in run["recall"]["notes"]] == ["a"] + + +def test_an_all_empty_run_still_reports_that_it_looked(): + blocks = [(0, {"type": "tool", "tool": "Bash", "status": "done", + "recall": {"status": RECALL_EMPTY}})] + run = _compact_activity_run(blocks, message_index=0) + assert run["recall"] == {"status": RECALL_EMPTY, "notes": []} + + +def test_a_successful_empty_lookup_outranks_a_failed_probe_in_the_same_run(): + blocks = [ + (0, {"type": "tool", "tool": "Bash", "status": "done", + "recall": {"status": RECALL_FAILED}}), + (1, {"type": "tool", "tool": "Bash", "status": "done", + "recall": {"status": RECALL_EMPTY}}), + ] + run = _compact_activity_run(blocks, message_index=0) + assert run["recall"] == {"status": RECALL_EMPTY, "notes": []} diff --git a/backend/tests/test_platform_update.py b/backend/tests/test_platform_update.py index 8d6073cf9..81e34d61d 100644 --- a/backend/tests/test_platform_update.py +++ b/backend/tests/test_platform_update.py @@ -1,20 +1,21 @@ """Clone-native platform reconcile — the git plumbing that fetches origin and -rebases the local edits onto the new version without ever losing them or serving -a broken tree. +merges it once with local edits without ever losing them or serving a broken +tree. These drive ``platform_update.reconcile_clone`` against throwaway repos in ``tmp_path``: a bare ``origin`` repo, a ``platform`` clone of it (mirroring the entrypoint bootstrap: local ``main`` + an ``upstream`` marker branch at HEAD), and the module's ``/data`` flag paths monkeypatched into ``tmp_path`` so no real platform tree is touched. Each platform tree carries a trivially-importable -``backend/app`` package so the post-rebase import probe (a real ``import +``backend/app`` package so the post-merge import probe (a real ``import app.main`` subprocess) exercises for real. -The load-bearing cases: a clean fast-forward advances the served tree; a disjoint -local edit is preserved by a rebase; a same-line conflict aborts and serves the -OLD code; a text-clean rebase whose result fails to import rolls back to the old -code; an offline fetch keeps serving unchanged; and a crash-interrupted rebase is -aborted on the next pass. +The load-bearing cases: a clean fast-forward advances the served tree; a +disjoint local edit is preserved with its original commit identity; all net +conflicts surface in one merge; a text-clean merge whose result fails to import +rolls back to the old code; an offline fetch keeps serving unchanged; and a +crash-interrupted merge is aborted on the next pass. A legacy interrupted rebase +is still cleaned up because an update can cross this implementation boundary. """ import subprocess @@ -38,7 +39,7 @@ def _git(cwd: Path, *args: str, check: bool = True) -> subprocess.CompletedProce # A trivially-importable backend so the import probe (`import app.main` with cwd # repo/backend) runs for real. `main.py` imports the sibling `foo` module so a -# test can delete `foo` upstream to make a text-clean rebase import-broken. +# test can delete `foo` upstream to make a text-clean merge import-broken. _MAIN_PY = "import app.foo\n\nVALUE = app.foo.VALUE\nLINE_A = 1\nLINE_B = 2\nLINE_C = 3\n" _FOO_PY = "VALUE = 'foo'\n" @@ -182,13 +183,13 @@ def fail_unshallow(_repo): assert _served_sha(platform) == new -# --- V-B2: disjoint local edit preserved via rebase ------------------------- +# --- V-B2: disjoint local edit preserved via one merge ---------------------- def test_local_edit_preserved_across_update(clone_env): origin, platform = clone_env - _local_commit(platform, edits={"backend/app/main.py": + local = _local_commit(platform, edits={"backend/app/main.py": _MAIN_PY.replace("LINE_A = 1", "LINE_A = 111")}) - _advance_origin(origin, edits={"backend/app/main.py": + target = _advance_origin(origin, edits={"backend/app/main.py": _MAIN_PY.replace("LINE_C = 3", "LINE_C = 333")}) res = pu.reconcile_clone(platform, at_boot=True) @@ -197,11 +198,23 @@ def test_local_edit_preserved_across_update(clone_env): served = (platform / "backend/app/main.py").read_text() assert "LINE_A = 111" in served # local edit assert "LINE_C = 333" in served # upstream edit + # A single merge preserves the original local commit instead of rewriting it + # through a per-commit rebase. Both complete histories are explicit parents. + parents = _git( + platform, "show", "-s", "--format=%P", res.new_sha, + ).stdout.split() + assert parents == [local, target] + assert _git( + platform, "merge-base", "--is-ancestor", local, res.new_sha, + ).returncode == 0 + assert _git( + platform, "merge-base", "--is-ancestor", target, res.new_sha, + ).returncode == 0 assert not pu.CONFLICT_FLAG.exists() # --- regression: a drifted upstream marker never triggers a data-losing -# fast-forward. The ff-vs-rebase choice is decided by ANCESTRY, not the upstream +# fast-forward. The ff-vs-merge choice is decided by ANCESTRY, not the upstream # marker, so committed local edits survive even when the marker is set to the # exact value that would have made the old marker-gated `reset --hard target` # discard them. This is the headline data-safety invariant of the fix. --------- @@ -221,7 +234,7 @@ def test_drifted_upstream_marker_never_discards_local_commits(clone_env): res = pu.reconcile_clone(platform, at_boot=True) - # local is NOT an ancestor of target, so ancestry forces a REBASE (not a reset) + # local is NOT an ancestor of target, so ancestry forces a MERGE (not a reset) # and BOTH survive — the local commit is not discarded despite the bad marker. assert res.status == "updated" served = (platform / "backend/app/main.py").read_text() @@ -244,10 +257,10 @@ def test_conflict_serves_old_and_flags(clone_env): res = pu.reconcile_clone(platform, at_boot=True) assert res.status == "conflict" - # Served tree is the pre-reconcile local code, intact; no half-rebase left. + # Served tree is the pre-reconcile local code, intact; no half-merge left. assert _served_sha(platform) == pre assert "LINE_A = 'LOCAL'" in (platform / "backend/app/main.py").read_text() - assert not pu._rebase_in_progress(platform) + assert not pu._reconcile_in_progress(platform) assert pu.CONFLICT_FLAG.exists() assert any("main.py" in p for p in res.conflict_paths) status = pu.platform_status(platform) @@ -255,11 +268,69 @@ def test_conflict_serves_old_and_flags(clone_env): assert any("main.py" in p for p in status["conflict_paths"]) -# --- V-B4: import-broken text-clean rebase -> rollback ---------------------- +def test_diverged_update_surfaces_all_net_conflicts_together(clone_env): + origin, platform = clone_env + _local_commit(platform, edits={"backend/app/main.py": + _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'LOCAL'")}) + _local_commit(platform, edits={"backend/app/foo.py": "VALUE = 'LOCAL'\n"}) + _advance_origin(origin, edits={ + "backend/app/main.py": _MAIN_PY.replace( + "LINE_A = 1", "LINE_A = 'UPSTREAM'", + ), + "backend/app/foo.py": "VALUE = 'UPSTREAM'\n", + }) + + res = pu.reconcile_clone(platform, at_boot=True) -def test_import_broken_rebase_rolls_back(clone_env): + # A per-commit rebase stopped on main.py, then made the resolver discover the + # foo.py conflict only after continuing. The net merge reports both at once. + assert res.status == "conflict" + assert set(res.conflict_paths) == { + "backend/app/main.py", "backend/app/foo.py", + } + assert not pu._reconcile_in_progress(platform) + + +@pytest.mark.parametrize( + ("merge_rc", "expected_error"), + [ + (pu._MERGE_TIMEOUT, "merge_timeout"), + (2, "merge_failed"), + ], +) +def test_non_conflict_merge_failure_serves_old_without_a_resolver_flag( + clone_env, monkeypatch, merge_rc, expected_error, +): origin, platform = clone_env - # A disjoint local edit (so the rebase is text-clean), while upstream DELETES + pre = _local_commit(platform, edits={"backend/app/main.py": + _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'LOCAL'")}) + target = _advance_origin(origin, edits={"backend/app/main.py": + _MAIN_PY.replace("LINE_C = 3", "LINE_C = 'UPSTREAM'")}) + # A previous attempt may have left either durable review state behind. A + # timeout or git failure with no unmerged paths has nothing a resolver can + # act on, so returning "error" while preserving one of these flags would make + # the next status read lie about what just happened. + pu._write_conflict_flag("b" * 40, ["backend/app/old.py"]) + pu._write_rolled_back_flag("c" * 40, "old import failure") + monkeypatch.setattr(pu, "_merge_target", lambda _repo, _target: merge_rc) + + res = pu.reconcile_clone(platform, at_boot=True) + + assert res.status == "error" + assert res.error == expected_error + assert res.target_sha == target + assert _served_sha(platform) == pre + assert not pu._reconcile_in_progress(platform) + assert not pu.CONFLICT_FLAG.exists() + assert not pu.ROLLED_BACK_FLAG.exists() + assert not pu.RECONCILE_PRE_FLAG.exists() + + +# --- V-B4: import-broken text-clean merge -> rollback ----------------------- + +def test_import_broken_merge_rolls_back(clone_env): + origin, platform = clone_env + # A disjoint local edit (so the merge is text-clean), while upstream DELETES # foo.py — which main.py still imports. Textually clean, import-broken. pre = _local_commit(platform, edits={"backend/app/main.py": _MAIN_PY + "LOCAL = 'kept'\n"}) @@ -335,29 +406,29 @@ def test_detached_head_uncommitted_edit_survives_reconcile(clone_env): assert "LINE_C = 1001" in served -# --- crash-safety: a stale in-progress rebase is aborted -------------------- +# --- crash-safety: interrupted merge + legacy rebase are aborted ------------ -def test_stale_rebase_aborted_on_next_pass(clone_env): +def test_stale_merge_aborted_on_next_pass(clone_env): origin, platform = clone_env - # Force a real conflict and leave the rebase in progress (no abort), mirroring - # a crash mid-rebase. + # Force a real conflict and leave the merge in progress (no abort), mirroring + # a crash mid-merge. pre = _local_commit(platform, edits={"backend/app/main.py": _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'LOCAL'")}) new = _advance_origin(origin, edits={"backend/app/main.py": _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'UPSTREAM'")}) _git(platform, "fetch", "-q", "origin") - rc = _git(platform, "rebase", new, "main", check=False).returncode - assert rc != 0 and pu._rebase_in_progress(platform) # left mid-rebase + rc = _git(platform, "merge", "--no-ff", new, check=False).returncode + assert rc != 0 and pu._merge_in_progress(platform) # left mid-merge - # Next reconcile must abort the stale rebase FIRST, then reconcile cleanly + # Next reconcile must abort the stale merge FIRST, then reconcile cleanly # (here: re-conflict and serve old — the point is it does not wedge or corrupt). res = pu.reconcile_clone(platform, at_boot=True) assert res.status == "conflict" assert _served_sha(platform) == pre - assert not pu._rebase_in_progress(platform) + assert not pu._reconcile_in_progress(platform) -def test_boot_guard_aborts_interrupted_rebase_before_serving(clone_env): +def test_boot_guard_aborts_interrupted_merge_before_serving(clone_env): origin, platform = clone_env pre = _local_commit(platform, edits={"backend/app/main.py": _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'LOCAL'")}) @@ -365,21 +436,38 @@ def test_boot_guard_aborts_interrupted_rebase_before_serving(clone_env): _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'UPSTREAM'")}) _git(platform, "fetch", "-q", "origin") pu._write_reconcile_pre(pre) - rc = _git(platform, "rebase", new, "main", check=False).returncode - assert rc != 0 and pu._rebase_in_progress(platform) + rc = _git(platform, "merge", "--no-ff", new, check=False).returncode + assert rc != 0 and pu._merge_in_progress(platform) assert "<<<<<<<" in (platform / "backend/app/main.py").read_text() summary = pu.boot_guard_clean_served_tree(platform) assert summary.startswith("boot_guard[reset]") assert _served_sha(platform) == pre - assert not pu._rebase_in_progress(platform) + assert not pu._reconcile_in_progress(platform) assert "<<<<<<<" not in (platform / "backend/app/main.py").read_text() ok, err = pu._import_probe(platform) assert ok, err assert not pu.RECONCILE_PRE_FLAG.exists() +def test_legacy_stale_rebase_aborted_on_next_pass(clone_env): + origin, platform = clone_env + pre = _local_commit(platform, edits={"backend/app/main.py": + _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'LOCAL'")}) + new = _advance_origin(origin, edits={"backend/app/main.py": + _MAIN_PY.replace("LINE_A = 1", "LINE_A = 'UPSTREAM'")}) + _git(platform, "fetch", "-q", "origin") + rc = _git(platform, "rebase", new, "main", check=False).returncode + assert rc != 0 and pu._rebase_in_progress(platform) + + res = pu.reconcile_clone(platform, at_boot=True) + + assert res.status == "conflict" + assert _served_sha(platform) == pre + assert not pu._reconcile_in_progress(platform) + + def test_boot_guard_sync_propagates_failure(monkeypatch): """The final boot gate must fail closed; callers need a non-zero process, not an error-looking success string that the shell can accidentally ignore.""" @@ -855,8 +943,8 @@ async def test_platform_conflict_resolver_chat_is_click_gated( pu._write_conflict_flag(target, ["backend/app/main.py"]) calls = [] - async def fake_spawn(db, paths): - calls.append((db, paths)) + async def fake_spawn(db, paths, target_sha): + calls.append((db, paths, target_sha)) return { "chat_id": "resolver-chat", "created": True, @@ -873,13 +961,30 @@ async def fake_spawn(db, paths): "created": True, "started": True, } - assert calls == [(db, ["backend/app/main.py"])] + assert calls == [(db, ["backend/app/main.py"], target)] flag = pu._read_conflict_flag() assert flag["upstream"] == target assert flag["paths"] == ["backend/app/main.py"] assert flag["chat_id"] == "resolver-chat" +def test_platform_conflict_resolver_message_pins_reviewed_target(): + target = "a" * 40 + + content = pu._platform_conflict_resolver_message( + target, + ["backend/app/main.py", "frontend/src/App.jsx"], + ) + + assert f"merge --no-ff {target}" in content + assert "merge --no-ff origin/main" not in content + assert "backend/app/main.py, frontend/src/App.jsx" in content + # The resolver runs in a headless shell where `git merge --continue` opens an + # editor and hangs/errors; it must finish non-interactively instead. + assert "commit --no-edit" in content + assert "merge --continue" not in content + + def test_status_restart_needed_when_disk_head_changed_after_boot(clone_env): origin, platform = clone_env served = _served_sha(platform) diff --git a/backend/tests/test_runner_registry_integration.py b/backend/tests/test_runner_registry_integration.py index 47066af94..e3ea6cd0b 100644 --- a/backend/tests/test_runner_registry_integration.py +++ b/backend/tests/test_runner_registry_integration.py @@ -152,6 +152,7 @@ async def thread_start(self, *_args, **_kwargs): "AsyncCodex": FakeAsyncCodex, "CodexConfig": lambda **kwargs: SimpleNamespace(**kwargs), "CodexRpcError": RuntimeError, + "TransportClosedError": type("TransportClosedError", (Exception,), {}), "CommandExecutionOutputDeltaNotification": type( "CommandExecutionOutputDeltaNotification", (), {} ), diff --git a/backend/tests/test_tool_output_excerpt.py b/backend/tests/test_tool_output_excerpt.py index 58c263144..bee70ab60 100644 --- a/backend/tests/test_tool_output_excerpt.py +++ b/backend/tests/test_tool_output_excerpt.py @@ -9,6 +9,7 @@ TOOL_OUTPUT_HEAD, TOOL_OUTPUT_INLINE_THRESHOLD, excerpt_tool_output, + tool_output_exit_code, ) @@ -38,6 +39,12 @@ def test_bash_failure_head_and_exit_code_survive_truncation(): assert full_len == len(content) +def test_small_tool_outputs_expose_the_same_typed_exit_code(): + assert tool_output_exit_code("Exit code 7\nfailed") == 7 + assert tool_output_exit_code(json.dumps({"stderr": "x", "exit_code": 2})) == 2 + assert tool_output_exit_code("success") is None + + def test_json_envelope_stays_valid_json_with_exit_code_intact(): envelope = { "stdout": "S" * 200000, diff --git a/backend/tests/test_tool_output_stash.py b/backend/tests/test_tool_output_stash.py index 214cd1664..c40094668 100644 --- a/backend/tests/test_tool_output_stash.py +++ b/backend/tests/test_tool_output_stash.py @@ -4,6 +4,7 @@ GET /tool-output/{tool_use_id} endpoint serves a bounded expansion preview and the exact text on explicit copy. Also covers the reducer carrying tool identity + truncation metadata onto the persisted block.""" +import json import uuid from sqlalchemy import event as sqlalchemy_event @@ -424,6 +425,49 @@ def test_sink_reduces_large_tagged_output_and_stashes_full(db): assert row is not None and row.output == big +def test_sink_keeps_a_runner_supplied_exit_code_on_large_plain_output(db): + sink = _sink() + sink.publish({ + "type": "tool_start", "tool": "Bash", "input": "run", + "tool_use_id": "tu_typed", + }) + big = "plain output\n" + ("x" * (TOOL_OUTPUT_INLINE_THRESHOLD + 100)) + event = { + "type": "tool_output", "content": big, "tool_use_id": "tu_typed", + "output_exit_code": 7, + } + + sink.publish(event) + + assert event["output_truncated"] is True + assert event["output_exit_code"] == 7 + assert sink.assistant_blocks[-1]["output_exit_code"] == 7 + + +def test_sink_parses_a_large_json_envelope_once(monkeypatch, db): + import app.events as events + + sink = _sink() + big = json.dumps({ + "stdout": "x" * (TOOL_OUTPUT_INLINE_THRESHOLD + 100), + "exit_code": 3, + }) + loads = events.json.loads + calls = 0 + + def counting_loads(value): + nonlocal calls + calls += 1 + return loads(value) + + monkeypatch.setattr(events.json, "loads", counting_loads) + event = {"type": "tool_output", "content": big, "tool_use_id": "tu_json"} + sink.publish(event) + + assert calls == 1 + assert event["output_exit_code"] == 3 + + def test_sink_passes_through_small_output(db): sink = _sink() small = "ok" diff --git a/frontend/src/components/ChatView/ActiveAssistantSurface.jsx b/frontend/src/components/ChatView/ActiveAssistantSurface.jsx index 7828949c5..867e4b5c3 100644 --- a/frontend/src/components/ChatView/ActiveAssistantSurface.jsx +++ b/frontend/src/components/ChatView/ActiveAssistantSurface.jsx @@ -32,6 +32,8 @@ function ActiveAssistantSurface({ onAutoResumeChange, submissionBlocked, liveQuestionId, + pendingQuestionRef, + resumeCardRef, isStreaming, }) { const msg = useMemo(() => { @@ -63,6 +65,8 @@ function ActiveAssistantSurface({ onAutoResumeChange={onAutoResumeChange} submissionBlocked={submissionBlocked} liveQuestionId={liveQuestionId} + pendingQuestionRef={pendingQuestionRef} + resumeCardRef={resumeCardRef} isStreaming={isStreaming} /> ) diff --git a/frontend/src/components/ChatView/ChatView.css b/frontend/src/components/ChatView/ChatView.css index c24a6a020..3e5b91800 100644 --- a/frontend/src/components/ChatView/ChatView.css +++ b/frontend/src/components/ChatView/ChatView.css @@ -978,6 +978,51 @@ line-height: 1; } +/* A recalled note reads as the same KIND of citation as a web source, so it + shares the chip entirely and diverges only in its mark: a glyph rather than + a domain letter, tinted to separate "remembered" from "read on the web" + without becoming a second visual language. */ +.chat__source-glyph { + width: 13px; + height: 13px; +} + +/* A note title is prose, not a domain, so it needs more room than a host chip + and is the part that should absorb any shortfall. */ +.chat__source-chip--memory { + max-width: 320px; +} + +.chat__source-chip--memory .chat__source-icon { + color: color-mix(in srgb, var(--accent) 62%, var(--text)); + background: color-mix(in srgb, var(--accent-dim) 52%, var(--surface2)); +} + +/* "Memory" is a fixed six-character label; letting it share the shrink budget + with the title collapses it to "Me…" and says nothing. */ +.chat__source-chip--memory .chat__source-host { + flex: 0 0 auto; +} + +/* An empty lookup is a fact about the answer, not a destination — so it is + stated plainly and styled back, with no hover or pointer affordance. */ +.chat__source-chip--quiet { + max-width: none; + color: var(--muted); /* CONTRACT: low-contrast text on opaque fill */ + background: color-mix(in srgb, var(--surface2) 34%, var(--surface)); + border-style: dashed; +} + +.chat__source-chip--quiet .chat__source-icon { + color: var(--muted); /* CONTRACT: low-contrast text on opaque fill */ + background: color-mix(in srgb, var(--surface2) 60%, var(--surface)); + border-color: transparent; +} + +.chat__source-chip--quiet .chat__source-title { + font-style: italic; +} + .chat__source-copy { display: flex; align-items: baseline; diff --git a/frontend/src/components/ChatView/ChatView.jsx b/frontend/src/components/ChatView/ChatView.jsx index 92127323e..88afb7d0e 100644 --- a/frontend/src/components/ChatView/ChatView.jsx +++ b/frontend/src/components/ChatView/ChatView.jsx @@ -23,7 +23,7 @@ import { getOnlineSnapshot } from '../../lib/connectivityStore.js' import useSystemEventStream from '../../hooks/useSystemEventStream.js' import usePendingQueue from './hooks/usePendingQueue.js' import useBridgePartial from './hooks/useBridgePartial.js' -import useOffscreenNudge from './hooks/useOffscreenNudge.js' +import useOffscreenNudge, { useNudgeTargetRef } from './hooks/useOffscreenNudge.js' import ChatInputBar from './ChatInputBar.jsx' import { hasSendablePayload } from './composerSubmission.js' import AgentContextInspector from './AgentContextInspector.jsx' @@ -421,7 +421,8 @@ export default function ChatView({ // The pending-question and resume "tap to jump to it" nudges each track // whether their card is scrolled out of the viewport. Both use one shared // observer hook (useOffscreenNudge, below); their booleans are computed near - // hasPendingQuestion / hasPendingResume where the card finders live. + // hasPendingQuestion / hasPendingResume, alongside the callback refs the + // cards themselves use to publish the node being observed. const [showInspector, setShowInspector] = useState(false) const [showSummary, setShowSummary] = useState(false) const [visibleTimestampKey, setVisibleTimestampKey] = useState(null) @@ -888,7 +889,14 @@ export default function ChatView({ // layout-derived floor and re-applies the active mode under the reader gate // (design §2). Skipped entirely for single-pane chats (paneContentHeight // null) so today's resize behavior is untouched. - useEffect(() => { + // + // Must be a layout effect: this is the only automatic scroll write in the + // controller that would otherwise run after paint. Every other one is + // pre-paint (syncLayout in a layout effect, the tail follow in a + // ResizeObserver callback, settleStreamingPin in rAF), and running this one + // post-paint shows the reader a frame at the old scroll position before the + // correction lands — visible as a jump when pane geometry changes. + useLayoutEffect(() => { if (paneContentHeight != null) paneResized(paneContentHeight) }, [paneContentHeight, paneResized]) @@ -1262,13 +1270,18 @@ export default function ChatView({ }, onLiveQuestion: setLiveQuestionId, onSteeredIntoTurn: ({ ts, content, messages: steeredBatch }) => { - // A send was injected mid-turn into a live turn (steering — fired for - // both providers when Stop is pressed with a queued message). The - // backend seals the assistant text streamed so far, persists the user - // message, and then continues the assistant after that boundary. Mirror - // that exact shape locally: first promote the current live stream - // segment into `messages`, then append the steered user row, then let - // future text deltas build a fresh streaming assistant block. + // The steer's transcript split has COMMITTED (fired for both providers, + // including when Stop is pressed with a queued message): the backend has + // sealed the assistant text streamed up to the split, persisted the user + // message after it, and reset for the continuation. Mirror that exact + // shape locally: first promote the current live stream segment into + // `messages`, then append the steered user row, then let future text + // deltas build a fresh streaming assistant block. + // + // The split is the route's own write for Codex and the runner's seal for + // Claude, so this event always arrives AFTER the last block belonging to + // the sealed segment. That ordering is what makes promoting the live + // stream here correct; it is not a guess about where the server cut. // // It still follows the one visible-row scroll rule. Automatic queue // promotion keeps the original submit snapshot; an explicit fast-forward @@ -1320,6 +1333,14 @@ export default function ChatView({ // instead of blindly appending: if a fetch/replay already committed the // post-steer assistant row, the steered user still belongs before it. commitMessages(prev => insertMessageBatchByTs(prev, steeredMessages)) + // The rows have now genuinely left `chat.pending_messages`, so retire the + // tray entries the deferred-cut window kept visible. A no-op on the + // route-split (Codex) path, where the send's own 202 already dropped + // them — the cut is simply the one place that owns the hand-off. + for (const msg of steeredMessages) { + const cid = cidOf(msg) + if (cid != null) pendingQueue.cancelByCid(cid) + } steerPinIntentRef.current = null }, }) @@ -2214,14 +2235,48 @@ export default function ChatView({ await handleSteerOneRef.current?.(cid) } } - // Mid-turn steer: the backend delivered the send into the live - // provider turn and persisted it in the transcript. The - // `steered_into_turn` SSE event (handled in useStreamConnection's - // onSteeredIntoTurn) renders the message inline, so drop the - // optimistic queued-tray entry here — it never queued. + // Mid-turn steer: the backend delivered the send into the live provider + // turn. Where the row LIVES right now is what `cut_deferred` states. if (result?.status === 'steered') { - pendingQueue.cancelByCid(queuedMsg.cid) - forgetQueuedPinIntent({ cid: queuedMsg.cid }) + if (result.cut_deferred) { + // Claude: the transcript split waits for the runner's interrupt + // boundary, so the row is STILL queued server-side and its tray + // entry stays — dropping it here left the owner's message with + // nowhere to render for the whole deferred window. Resolve THIS + // send's own row and nothing else: confirm it by cid against the + // server's echoed entry (which carries the durable ts fast-forward + // needs). + // + // Not `hydrate(result.pending_messages)`: that list is a snapshot + // taken at steer time, and a wholesale reconcile against it is + // wrong in both directions. It would DROP a row queued and + // confirmed while this 202 was in flight (absent from the snapshot, + // no longer in-flight), and it would RESURRECT this row if the + // runner's cut landed first (the cut retires the tray entry, then + // the snapshot puts it back while it is already inline). Confirming + // one cid is a no-op once the cut has retired it, so the two can + // land in either order. + const serverRow = Array.isArray(result.pending_messages) + ? result.pending_messages.find(m => cidOf(m) === queuedMsg.cid) + : null + pendingQueue.confirmQueued(queuedMsg.cid, { + ts: serverRow?.ts ?? queuedMsg.ts, + position: serverRow?.position, + serverMsg: serverRow, + }) + // The pin intent lives on in `inlineSteerPinIntentRef` (set at + // submit), which is what `onSteeredIntoTurn` reads at the cut. Its + // `takeQueuedPinIntent` fallback is short-circuited by that ref, so + // without this the map entry would never be taken and would leak + // for the life of the mounted chat. + forgetQueuedPinIntent({ cid: queuedMsg.cid }) + } else { + // Codex: the split already ran at the route, so the row is in the + // transcript and `steered_into_turn` (already sent) renders it + // inline — drop the optimistic queued-tray entry, it never queued. + pendingQueue.cancelByCid(queuedMsg.cid) + forgetQueuedPinIntent({ cid: queuedMsg.cid }) + } } // Race: server said "started" though we expected queued. if (result?.status === 'started') { @@ -2270,7 +2325,10 @@ export default function ChatView({ } // Invariant: every observable queue-path status must resolve // the optimistic entry's in-flight flag. queued/steered/started - // each clear it above (confirmQueued / cancelByCid). Any + // each clear it above, unconditionally, via confirmQueued or + // cancelByCid — including BOTH steered branches (confirmQueued when + // the cut is deferred, cancelByCid when the route already split), so + // no response shape can slip through leaving the mark set. Any // other status — e.g. streamSend's `not_steered` — leaves the // entry as an ordinary queued row, so clear the flag here or it // leaks forever and a later hydrate would wrongly preserve it. @@ -2984,10 +3042,12 @@ export default function ChatView({ // or waiting for turn-end (the default queue drain). Mirrors handleStop's // structure — re-entry guard, snapshot-before-await — but never // interrupts the running turn. The backend force-steers (bypassing the - // steer_enabled opt-in) for BOTH providers; on success the steered - // message lands in the transcript and renders inline via the - // `steered_into_turn` SSE event (onSteeredIntoTurn above), so we just - // drop those rows from the local tray. + // steer_enabled opt-in) for BOTH providers; the rows render inline when the + // `steered_into_turn` SSE event reports the transcript split (onSteeredIntoTurn + // above), and THAT is when they leave the local tray. Codex splits at the + // route, so the split is already done when the POST resolves; Claude splits at + // its next content-block boundary, so its 202 comes back `cut_deferred` and + // the rows stay in the tray until the cut lands. // The shared force-steer core: given serverTs-CONFIRMED queue rows (in // queue order), optimistically hide them, POST one force_steer selecting // them by cid, and reconcile or restore. Restore re-hydrates the full @@ -3032,9 +3092,10 @@ export default function ChatView({ let queueAfterOptimisticPromote = null function restoreOptimisticSteerQueue() { // If another path touched the queue while the POST was in flight - // (notably the natural turn-end drain), every pendingQueue mutation - // assigns a fresh array. In that case the other path won the race, - // so restoring our stale snapshot would resurrect duplicate chips. + // (notably the natural turn-end drain, or a deferred steer's own cut + // arriving before its 202 resolves), every pendingQueue mutation assigns + // a fresh array. In that case the other path won the race, so restoring + // our stale snapshot would resurrect duplicate chips. if ( queueAfterOptimisticPromote !== null && pendingQueue.pendingMessagesRef.current === queueAfterOptimisticPromote @@ -3067,6 +3128,13 @@ export default function ChatView({ // visible "down, then up" fast-forward jump. Hide only the confirmed // rows this request is steering; restore the snapshot below if the // backend says the turn was not steered. + // + // A `cut_deferred` steer (Claude) puts these rows straight back below: + // they are still queued server-side until the runner's seal, and the pin + // is armed at the cut, not here, so this hide buys nothing there. We + // cannot know which case it is before the POST resolves, and the response + // restores them in the same round-trip, so the deferred path just doesn't + // get the pre-pin hide. pendingQueue.promoteManyByCid(consumePendingCids) queueAfterOptimisticPromote = pendingQueue.pendingMessagesRef.current const result = await streamSend(content, attachments, { @@ -3080,13 +3148,24 @@ export default function ChatView({ })), }) if (result?.status === 'steered') { - // The steered rows now render inline (onSteeredIntoTurn promotes - // them from the SSE event + transcript). Drop them from the local - // tray. Reconcile against the server's authoritative remaining - // queue when present, else remove exactly the steered cids. - if (Array.isArray(result.pending_messages)) { + if (result.cut_deferred) { + // Claude: the split waits for the runner's interrupt boundary, so + // these rows are still queued server-side and the optimistic hide + // above has to come back. Undo it with the SAME identity-guarded + // restore the not_steered path uses, not with the response's echoed + // queue: that list is a snapshot from steer time, so re-adding from + // it would resurrect a row the cut had already retired (the cut can + // land before this 202 resolves) and would drop a row queued while + // the POST was in flight. The guard makes the restore a no-op exactly + // when something else — the cut included — already owns the queue. + restoreOptimisticSteerQueue() + } else if (Array.isArray(result.pending_messages)) { + // Codex: the route already split, so the echoed queue no longer holds + // these rows and onSteeredIntoTurn has rendered them inline. pendingQueue.hydrate(result.pending_messages) } else { + // A backend that echoes no queue at all: remove exactly the steered + // cids. for (const c of consumePendingCids) pendingQueue.cancelByCid(c) } forgetQueuedPinIntent({ cidList: consumePendingCids }) @@ -3111,10 +3190,12 @@ export default function ChatView({ // or waiting for turn-end (the default queue drain). Mirrors handleStop's // structure — re-entry guard, snapshot-before-await — but never // interrupts the running turn. The backend force-steers (bypassing the - // steer_enabled opt-in) for BOTH providers; on success the steered - // message lands in the transcript and renders inline via the - // `steered_into_turn` SSE event (onSteeredIntoTurn above), so we just - // drop those rows from the local tray. + // steer_enabled opt-in) for BOTH providers; the rows render inline when the + // `steered_into_turn` SSE event reports the transcript split (onSteeredIntoTurn + // above), and THAT is when they leave the local tray. Codex splits at the + // route, so the split is already done when the POST resolves; Claude splits at + // its next content-block boundary, so its 202 comes back `cut_deferred` and + // the rows stay in the tray until the cut lands. async function handleSteer() { if (handlingSteerRef.current) return handlingSteerRef.current = true @@ -3518,30 +3599,30 @@ export default function ChatView({ } }, [pendingLimitResetAt]) - // Visibility of that card is a pure viewport question — an - // IntersectionObserver rooted at the scroll container is the signal, - // no scroll math and no interaction with the spacer machinery. The - // card's DOM node is stable across streaming ticks (keyed children), - // so the observer only needs re-binding when the rendering surface - // can change: pending-flag flips, stream↔messages promotion, or a - // messages commit. Both nudges share useOffscreenNudge (above). - // The LAST un-answered card is the pending one: it lives in the last - // assistant message or the streaming
  • . - const findPendingQuestionCard = () => - [...(scrollRef.current?.querySelectorAll('.qcard:not(.qcard--answered)') ?? [])].pop() + // Visibility of either card is a pure viewport question — an + // IntersectionObserver rooted at the scroll container is the signal, no + // scroll math and no interaction with the spacer machinery. ChatView must + // NOT look the card up: the pending question moves between two rendering + // surfaces (the live streaming
  • and the durable message row) at a moment + // ChatView cannot enumerate, and a lookup taken at bind time then observes a + // node React has since detached — with the turn parked on the answer nothing + // re-renders, so the cue would stick forever. Instead the element that IS + // the card publishes its node through these refs (see useNudgeTargetRef); + // both surfaces publish through the same channel, so the live→durable + // handoff reaches the observer as an ordinary node swap. + const [pendingQuestionEl, pendingQuestionRef] = useNudgeTargetRef() const pendingCardOffscreen = useOffscreenNudge( - scrollRef, hasPendingQuestion, findPendingQuestionCard, - [showActiveAssistantSurface, messages], + scrollRef, hasPendingQuestion, pendingQuestionEl, ) - // The resume card: only the tail resumable note renders `.chat__resume` - // (MsgContent gates the button on isLastMsg), so observing that button is - // enough to know the card's visibility; a tap on the nudge scrolls it in. - const findResumeCard = () => - [...(scrollRef.current?.querySelectorAll('.chat__resume') ?? [])].pop() + // The resume card publishes the same way, from the TAIL resumable note only + // — the same block tailResumableBlock arms the cue on. (MsgContent renders a + // Resume button on every resumable block of the last message, so the tail + // gate lives at the publication site; one shared ref can only hold one + // node.) A tap on the nudge scrolls that node in. + const [resumeCardEl, resumeCardRef] = useNudgeTargetRef() const resumeCardOffscreen = useOffscreenNudge( - scrollRef, hasPendingResume, findResumeCard, - [showActiveAssistantSurface, messages], + scrollRef, hasPendingResume, resumeCardEl, ) // The ONE active
  • carries this data-key for both DB and live payloads. @@ -3846,6 +3927,8 @@ export default function ChatView({ isLastMsg={isLastMsg} liveQuestionId={liveQuestionId} suppressedQuestionKeys={streamItemQuestionKeys} + pendingQuestionRef={pendingQuestionRef} + resumeCardRef={resumeCardRef} /> {msg.ts && ownerUserMessage && (