diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e3731f587..c56916ff0 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -869,6 +869,19 @@ All chat-domain mutations — transcript writes, run-markers, question rows, ans Streaming state is physically bounded: `PersistTranscript`/`PersistError` replace `Chat.live_assistant`, never the historical `Chat.messages` JSON blob. Read routes overlay that current assistant on immutable history. `QuestionCommit` merges the card into history before broadcast, `Finalize` performs the terminal merge and clears the live value, and startup reconciliation performs the same merge after a crash. This keeps one-second crash-resilient snapshots without quadratic transcript rewrites as chats grow. +Settled transcript reads have a separate bounded presentation contract. +`GET /api/chats/{id}?compact=1` keeps prose, cards, distinctive image-view +beats, and small collapsed activity metadata, but replaces each multi-step +thinking/tool run with an `activity` reference into the immutable stored +message. Repeated steps are bounded by activity variety rather than raw call +count. Only an explicit disclosure resolves that exact range through +`GET /api/chats/{id}/activity-detail`; the live assistant stays self-contained. +Mounted runtime reconciliation uses `GET /api/chats/{id}/runtime`, whose ORM +projection raiseloads every unrequested field so polling can never silently +decode `Chat.messages`. These are read projections, never a second persistence +format: provider context, recovery, export, and writer commands continue to +use the full transcript. + - **Commit-before-ack (strict paths):** the caller's `await` on `QuestionCommit`/`Finalize`/`AnswerQuestion`/`Barrier`/`DrainAndStop` doesn't unblock until the commit succeeds; `PersistTranscript` and `PersistError` are fire-and-forget (submitted without awaiting the ack). - **Questions commit-before-broadcast:** a question row is durable before its SSE push fires, so a reconnect's catch-up burst always finds it. - **Concurrency invariant:** ack `Future`s are NEVER resolved while a producer lock is held — collect `(ack, value)` under the lock, resolve after release — so even a synchronous done-callback that re-enters `submit()`/`stop()` can't deadlock. Do not move an ack resolution back inside a `with` block. diff --git a/backend/app/app_compile_contract.py b/backend/app/app_compile_contract.py index 7ef956ecc..dcde3ed63 100644 --- a/backend/app/app_compile_contract.py +++ b/backend/app/app_compile_contract.py @@ -154,6 +154,16 @@ def esbuild_command( "--format=esm", "--jsx=automatic", "--platform=browser", + # Mini-apps are runtime artifacts, not development builds. Without an + # explicit production define React selects its development branches and + # every app carries nearly a megabyte of validation/debug code that the + # opaque frame must copy, parse, and execute on every cold mount. + '--define:process.env.NODE_ENV="production"', + # Keep the one-module offline contract while reducing transfer, cache, + # parse, and evaluation cost. Preserve Function.name/Class.name for apps + # that use names in labels or diagnostics. + "--minify", + "--keep-names", f"--banner:js={COMPILED_RUNTIME_BANNER}", f"--inject:{runtime_inject_path()}", f"--alias:mobius-runtime={mobius_runtime_path()}", diff --git a/backend/app/app_preview.py b/backend/app/app_preview.py new file mode 100644 index 000000000..5fb2924ca --- /dev/null +++ b/backend/app/app_preview.py @@ -0,0 +1,96 @@ +"""Durable per-build acknowledgement for an app's owning-chat open button.""" + +from datetime import UTC, datetime + +from sqlalchemy import update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app import models + + +def naive_utc(value: datetime) -> datetime: + """Normalize an API datetime to the naive-UTC shape SQLite returns.""" + if value.tzinfo is not None: + return value.astimezone(UTC).replace(tzinfo=None) + return value + + +def _advance_existing( + db: Session, app_id: int, seen_updated_at: datetime, seen_as_final: bool, +) -> bool: + """Advance one row without letting an older acknowledgement move it back.""" + advanced = db.execute( + update(models.AppPreviewState) + .where( + models.AppPreviewState.app_id == app_id, + models.AppPreviewState.seen_updated_at < seen_updated_at, + ) + .values( + seen_updated_at=seen_updated_at, + seen_as_final=seen_as_final, + ) + ) + if advanced.rowcount: + return True + if seen_as_final: + promoted = db.execute( + update(models.AppPreviewState) + .where( + models.AppPreviewState.app_id == app_id, + models.AppPreviewState.seen_updated_at == seen_updated_at, + models.AppPreviewState.seen_as_final.is_(False), + ) + .values(seen_as_final=True) + ) + if promoted.rowcount: + return True + return db.get(models.AppPreviewState, app_id) is not None + + +def mark_seen( + db: Session, + app_id: int, + seen_updated_at: datetime, + *, + seen_as_final: bool, +) -> None: + """Acknowledge only the build the opening shell actually observed. + + An older request may arrive after a newer build was opened on another device. + The monotonic timestamp update keeps that late request from hiding or + downgrading the newer acknowledgement. + """ + seen_updated_at = naive_utc(seen_updated_at) + if _advance_existing(db, app_id, seen_updated_at, seen_as_final): + return + try: + with db.begin_nested(): + db.add(models.AppPreviewState( + app_id=app_id, + seen_updated_at=seen_updated_at, + seen_as_final=seen_as_final, + )) + db.flush() + except IntegrityError: + # Two devices acknowledged the first visible build concurrently. The + # savepoint preserves the outer request; replay the monotonic update. + _advance_existing(db, app_id, seen_updated_at, seen_as_final) + + +def annotate_apps(db: Session, apps: list[models.App]) -> list[models.App]: + """Attach response-only preview acknowledgement fields to app rows.""" + ids = [app.id for app in apps] + state_by_id = {} + if ids: + state_by_id = { + row.app_id: row + for row in db.query(models.AppPreviewState).filter( + models.AppPreviewState.app_id.in_(ids) + ).all() + } + for app in apps: + state = state_by_id.get(app.id) + app.preview_seen_updated_at = state.seen_updated_at if state else None + app.preview_seen_final = bool(state and state.seen_as_final) + return apps diff --git a/backend/app/chat.py b/backend/app/chat.py index 9f0bb7220..edda04e31 100644 --- a/backend/app/chat.py +++ b/backend/app/chat.py @@ -54,6 +54,7 @@ PersistError, PersistTranscript, QuestionCommit, + RecordRunMetrics, ResolvePark, RollbackAutoResume, StashThinkingTrace, @@ -1194,6 +1195,45 @@ async def _clear_run_status( ) +async def _record_run_metrics( + *, + chat_id: str, + run_token: str, + provider_session_id: str | None, + cost_usd: float | None, + usage: dict | None, +) -> None: + """Best-effort durable accounting for one provider run. + + Usage must not be able to turn an otherwise successful chat response into a + failed turn. The exact run identity keeps a delayed completion from + attributing counters to a successor, and the writer actor keeps this scalar + update ordered with the later terminal transition. + """ + if not chat_id or not run_token: + return + # A provider can legitimately omit usage (and Codex currently omits cost). + # With no accounting signal there is nothing to record; avoiding a no-op + # actor round-trip also preserves the runner's connection-release contract. + if usage is None and cost_usd in (None, 0): + return + try: + await _await_ack(get_writer().submit(RecordRunMetrics( + chat_id=chat_id, + run_token=run_token, + provider_session_id=provider_session_id, + cost_usd=cost_usd, + usage=usage, + ))) + except Exception: + _get_logger().warning( + "RecordRunMetrics did not persist chat_id=%s run_token=%s", + chat_id, + run_token, + exc_info=True, + ) + + async def _clear_run_status_strict( chat_id: str, run_token: str = "", @@ -2213,10 +2253,12 @@ async def _auto_resume_chat( # Share the queue lock with owner/app sends. The outer sweep check can # go stale while this task waits, so re-check global liveness, policy, # attribution, the exact latest park, and provider ownership at the - # actual claim point. + # actual claim point. Provider-limit retries are globally serial to + # avoid a reset storm. Planned-restart continuations are different: + # they are the exact, owner-opted set that was already live together + # before the restart, so each chat may reclaim its own slot + # independently. async with chat_queue.get_lock(chat_id): - if _any_chat_turn_active(): - return False with SessionLocal() as check_db: chat = check_db.query(models.Chat).filter( models.Chat.id == chat_id, @@ -2275,6 +2317,11 @@ async def _auto_resume_chat( resume_reason = ( "restart" if park.park_reason == "restart" else "usage_limit" ) + if ( + resume_reason != "restart" + and _any_chat_turn_active() + ): + return False if not mark_starting(chat_id): return False claimed = True @@ -2369,11 +2416,13 @@ async def sweep_reset_parks(db: Session) -> list[str]: failure cannot silently consume the promised continuation. The narrow post-promote SIGKILL boundary is documented on `_auto_resume_chat`. - A park whose chat was deleted resolves silently. - - Auto-resume is controlled per chat and STRICTLY SERIAL: at most one - enabled park starts per tick, and none while any turn is live anywhere. - A blocked enabled chat stays pending for a later tick, while notify-only - chats in the same due batch still resolve normally. App-attributed runs - and queues never auto-resume. + - Auto-resume is controlled per chat. Provider-limit retries are strictly + serial: at most one starts per tick, and none while any turn is live + anywhere. Planned-restart continuations reclaim the exact set that was + already live before the restart, so every eligible chat in the batch may + resume independently. A blocked enabled chat stays pending for a later + tick, while notify-only chats in the same due batch still resolve + normally. App-attributed runs and queues never auto-resume. Stands down while draining — a restart is in progress, and the fresh process's immediate sweep picks everything up. Never raises. @@ -2485,14 +2534,15 @@ def wants_auto_resume(chat, run) -> bool: and restart_authorized ) - auto_resume_started = False + limit_resume_started = False for run in due: chat_id = run.chat_id chat = chats.get(chat_id) chat_gone = chat is None or chat.deleted_at is not None auto_resume = wants_auto_resume(chat, run) - if auto_resume and ( - auto_resume_started or _any_chat_turn_active() + restart_auto_resume = auto_resume and run.park_reason == "restart" + if auto_resume and not restart_auto_resume and ( + limit_resume_started or _any_chat_turn_active() ): # Strictly-serial gate: a live turn (an earlier auto-resume, or the # owner's own send) must settle before this enabled park is processed. @@ -2550,16 +2600,18 @@ def wants_auto_resume(chat, run) -> bool: if prepared.get("notify"): notify_due(chat_id, run) - if _any_chat_turn_active(): + if not restart_auto_resume and _any_chat_turn_active(): # The notification or refresh window admitted another turn. Keep the # durable pending state so the next sweep retries instead of silently # dropping the promised continuation. continue - auto_resume_started = await _auto_resume_chat( + resume_started = await _auto_resume_chat( chat_id, park_token=run.id, ) - if auto_resume_started: + if resume_started: resolved.append(chat_id) + if not restart_auto_resume: + limit_resume_started = True continue # Notify-only/app/deleted path: resolve before the best-effort push so a @@ -3647,7 +3699,8 @@ async def _complete_turn( # legitimately-silent turn: a user Stop lands as stop_handoff_successor (or # disowns the generation above), a park sets limit_reached, an errored/refused # turn sets _last_error, and any real text/thinking/tool_use makes the blocks - # renderable. cost_usd is unusable (None for every run here) and not consulted. + # renderable. Provider usage/cost is accounting data, not proof of a reply, + # and is deliberately not consulted. lost_reply = ( we_own_gen and not stop_handoff_successor @@ -5144,6 +5197,14 @@ async def _run_chat_impl_with_db( ) new_session_id = runner_result.get("session_id") err = runner_result.get("error") + usage_metrics = runner_result.get("usage_metrics") + await _record_run_metrics( + chat_id=chat_id, + run_token=run_token or "", + provider_session_id=new_session_id or session_id, + cost_usd=runner_result.get("cost_usd"), + usage=usage_metrics, + ) if not err and new_session_id and chat_id: chat_obj = db.query(models.Chat).filter( models.Chat.id == chat_id @@ -5161,10 +5222,14 @@ async def _run_chat_impl_with_db( ) else: log.info( - "chat done chat_id=%s cost_usd=%.4f sdk=codex status=%s phase=%s", + "chat done chat_id=%s cost_usd=%.4f sdk=codex status=%s phase=%s " + "input_tokens=%s output_tokens=%s total_tokens=%s", chat_id, runner_result.get("cost_usd") or 0.0, runner_result.get("terminal_status"), runner_result.get("final_message_phase"), + (usage_metrics or {}).get("input_tokens"), + (usage_metrics or {}).get("output_tokens"), + (usage_metrics or {}).get("total_tokens"), ) except Exception as exc: log.exception("codex SDK turn failed chat_id=%s: %s", chat_id, exc) @@ -5266,6 +5331,14 @@ async def _run_chat_impl_with_db( ) new_session_id = runner_result.get("session_id") err = runner_result.get("error") + usage_metrics = runner_result.get("usage_metrics") + await _record_run_metrics( + chat_id=chat_id, + run_token=run_token or "", + provider_session_id=new_session_id or claude_session_id, + cost_usd=runner_result.get("cost_usd"), + usage=usage_metrics, + ) if not err and new_session_id and chat_id: chat_obj = db.query(models.Chat).filter( models.Chat.id == chat_id @@ -5277,8 +5350,12 @@ async def _run_chat_impl_with_db( log.error("claude SDK error chat_id=%s: %s", chat_id, err) else: log.info( - "chat done chat_id=%s cost_usd=%.4f sdk=claude", + "chat done chat_id=%s cost_usd=%.4f sdk=claude " + "input_tokens=%s output_tokens=%s total_tokens=%s", chat_id, runner_result.get("cost_usd") or 0.0, + (usage_metrics or {}).get("input_tokens"), + (usage_metrics or {}).get("output_tokens"), + (usage_metrics or {}).get("total_tokens"), ) except Exception as exc: log.exception("claude SDK turn failed chat_id=%s: %s", chat_id, exc) diff --git a/backend/app/chat_transcript.py b/backend/app/chat_transcript.py index fab141b03..4da520fbd 100644 --- a/backend/app/chat_transcript.py +++ b/backend/app/chat_transcript.py @@ -2,6 +2,17 @@ from __future__ import annotations +import re + + +_QUESTION_TOOLS = {"AskUserQuestion", "request_user_input"} +_IMAGE_PATH_RE = re.compile( + r"\.(?:avif|bmp|gif|heic|heif|jpe?g|png|svg|webp)(?:[?#].*)?$", + re.IGNORECASE, +) +_MAX_COMPACT_SOURCES = 24 +MAX_ACTIVITY_DETAIL_BLOCKS = 2000 + def materialized_messages(chat) -> list[dict]: """Return history with a visible in-flight assistant snapshot overlaid.""" @@ -101,3 +112,234 @@ def historical_tool_output_ids( continue ids.add(block["tool_use_id"]) return 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": + return False + raw = block.get("input") + if isinstance(raw, dict): + raw = raw.get("file_path") or raw.get("path") or "" + return isinstance(raw, str) and bool(_IMAGE_PATH_RE.search(raw)) + + +def _compact_activity_item(block: dict) -> dict: + """Return only the metadata needed to paint a collapsed activity line.""" + if block.get("type") == "thinking": + return { + "type": "thinking", + **( + {"thinking_id": block["thinking_id"]} + if isinstance(block.get("thinking_id"), str) + and block["thinking_id"] + else {} + ), + **( + {"duration_ms": block["duration_ms"]} + if isinstance(block.get("duration_ms"), (int, float)) + else {} + ), + } + + tool = { + "type": "tool", + "tool": block.get("tool") or "Tool", + # This projection never touches the live assistant. A stale persisted + # "running" flag belongs to an interrupted historical step, not current + # liveness, so normalize it at the read boundary. + "status": ( + "done" if block.get("status") == "running" + else block.get("status") or "done" + ), + } + for key in ("tool_use_id", "output_exit_code", "subagent"): + if key in block: + tool[key] = block[key] + # Read's path is the only input that affects the collapsed presentation: + # image reads are intentionally a distinctive beat. Keep it bounded; full + # 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] + return tool + + +def _compact_activity_entries( + blocks: list[tuple[int, dict]], +) -> list[dict]: + """Bound header metadata by activity variety, not raw step count. + + Repeated shell/edit loops are the pathological transcripts this projection + exists for. The collapsed label needs first-seen activity order and whether + an activity occurred once or repeatedly, so two entries per tool name are + sufficient. Keep every helper-bearing or failed entry because those facts + remain visible while collapsed. Thinking contributes one entry with the + run's total measured duration. + """ + entries: list[dict] = [] + tool_counts: dict[str, int] = {} + first_thinking_entry: dict | None = None + thinking_duration = 0.0 + has_thinking_duration = False + + for raw_index, block in blocks: + if block.get("type") == "thinking": + duration = block.get("duration_ms") + if isinstance(duration, (int, float)) and not isinstance(duration, bool): + thinking_duration += duration + has_thinking_duration = True + if first_thinking_entry is None: + first_thinking_entry = { + "item": _compact_activity_item(block), + "idx": raw_index, + } + entries.append(first_thinking_entry) + continue + + tool_name = block.get("tool") or "Tool" + occurrence = tool_counts.get(tool_name, 0) + 1 + tool_counts[tool_name] = occurrence + has_helpers = isinstance(block.get("subagent"), dict) and block["subagent"] + exit_code = block.get("output_exit_code") + failed = ( + isinstance(exit_code, (int, float)) + and not isinstance(exit_code, bool) + and exit_code != 0 + ) + if occurrence <= 2 or has_helpers or failed: + entries.append({ + "item": _compact_activity_item(block), + "idx": raw_index, + }) + + if first_thinking_entry is not None: + item = first_thinking_entry["item"] + if has_thinking_duration: + item["duration_ms"] = thinking_duration + else: + item.pop("duration_ms", None) + return entries + + +def _compact_activity_run( + blocks: list[tuple[int, dict]], + *, + message_index: int, +) -> dict: + sources: list[dict] = [] + seen_source_urls: set[str] = set() + for _, block in blocks: + for source in block.get("sources") or []: + if not isinstance(source, dict): + continue + url = source.get("url") + if not isinstance(url, str) or not url or url in seen_source_urls: + continue + seen_source_urls.add(url) + sources.append(source) + if len(sources) >= _MAX_COMPACT_SOURCES: + break + if len(sources) >= _MAX_COMPACT_SOURCES: + break + + start = blocks[0][0] + end = blocks[-1][0] + 1 + return { + "type": "activity", + "activity_id": f"{message_index}:{start}:{end}", + "message_index": message_index, + "start": start, + "end": end, + "entries": _compact_activity_entries(blocks), + "tool_count": sum( + block.get("type") == "tool" for _, block in blocks + ), + **({"sources": sources} if sources else {}), + } + + +def compact_messages_for_detail( + messages: list[dict], + *, + message_offset: int, + live_message: dict | None = None, +) -> list[dict]: + """Project settled activity runs into small, lazily expandable summaries. + + Stored ``Chat.messages`` remains the full source of truth. The normal chat + read only needs prose, cards, and the metadata that paints each collapsed + activity header. Expanding a header reads its original block range through + the activity-detail endpoint. + + Single activity entries stay inline: introducing a network boundary for one + ordinary tool/thought would cost more complexity than it saves. Distinctive + image-view beats also stay independent, preserving the transcript's visual + punctuation. Question-tool twins are omitted when the message already owns + the canonical question card, matching the frontend's historical repair. + """ + projected: list[dict] | None = None + for page_index, message in enumerate(messages): + if message is live_message or message.get("role") != "assistant": + continue + blocks = message.get("blocks") + if not isinstance(blocks, list) or len(blocks) < 2: + continue + + has_question = any( + isinstance(block, dict) and block.get("type") == "question" + for block in blocks + ) + next_blocks: list[dict] = [] + run: list[tuple[int, dict]] = [] + changed = False + + def flush() -> None: + nonlocal changed + while run: + chunk = run[:MAX_ACTIVITY_DETAIL_BLOCKS] + del run[:MAX_ACTIVITY_DETAIL_BLOCKS] + if len(chunk) >= 2: + next_blocks.append(_compact_activity_run( + chunk, + message_index=message_offset + page_index, + )) + changed = True + else: + next_blocks.extend(block for _, block in chunk) + run.clear() + + for raw_index, block in enumerate(blocks): + activity = ( + isinstance(block, dict) + and block.get("type") in {"tool", "thinking"} + ) + question_twin = ( + activity + and block.get("type") == "tool" + and has_question + and block.get("tool") in _QUESTION_TOOLS + ) + if question_twin: + flush() + changed = True + continue + if activity and not _distinctive_activity(block): + run.append((raw_index, block)) + continue + flush() + next_blocks.append(block) + flush() + + if not changed: + continue + if projected is None: + projected = list(messages) + next_message = dict(message) + next_message["blocks"] = next_blocks + # Assistant content duplicates its text blocks. Once blocks are present, + # copying and rendering already read those blocks, so the duplicate string + # only inflates parse/cache cost. + next_message.pop("content", None) + projected[page_index] = next_message + + return projected if projected is not None else messages diff --git a/backend/app/chat_writer.py b/backend/app/chat_writer.py index f56d4f090..5dc2b1a48 100644 --- a/backend/app/chat_writer.py +++ b/backend/app/chat_writer.py @@ -223,6 +223,23 @@ class PersistSessionId(_Command): session_id: str = "" +@dataclass +class RecordRunMetrics(_Command): + """Persist provider-neutral usage/cost counters on one ChatRun. + + This is separate from transcript Finalize because a stale/stop handoff may + intentionally skip transcript mutation while its own run row still deserves + the provider usage reported for it. Identity is the run token; this command + never reads or mutates Chat.messages. + """ + + chat_id: str = "" + run_token: str = "" + provider_session_id: str | None = None + cost_usd: float | None = None + usage: dict | None = None + + @dataclass class RecordAgentLifecycle(_Command): """Append one normalized helper lifecycle milestone. @@ -1352,6 +1369,8 @@ def _dispatch(self, db, cmd: _Command): return self._answer_question(db, cmd) if isinstance(cmd, PersistSessionId): return self._persist_session_id(db, cmd) + if isinstance(cmd, RecordRunMetrics): + return self._record_run_metrics(db, cmd) if isinstance(cmd, RecordAgentLifecycle): from app.agent_lifecycle import record_event return record_event(db, cmd.values) @@ -1542,6 +1561,45 @@ def _persist_session_id(self, db, cmd: PersistSessionId) -> bool: raise _PersistFailed("PersistSessionId did not persist") return True + def _record_run_metrics(self, db, cmd: RecordRunMetrics) -> bool: + """Attach provider usage/cost to the exact durable run row.""" + if not cmd.chat_id or not cmd.run_token: + return False + from app.models import ChatRun + + run = db.query(ChatRun).filter( + ChatRun.id == cmd.run_token, + ChatRun.chat_id == cmd.chat_id, + ).first() + if run is None: + raise _PersistFailed("RecordRunMetrics: run not found") + + if cmd.provider_session_id: + run.provider_session_id = cmd.provider_session_id + if cmd.cost_usd is not None: + run.cost_usd = float(cmd.cost_usd) + usage = copy.deepcopy(cmd.usage) if cmd.usage else None + if usage is not None: + run.usage_json = usage + for field_name in ( + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "reasoning_output_tokens", + "total_tokens", + "model_context_window", + ): + value = usage.get(field_name) + setattr( + run, + field_name, + int(value) if value is not None else None, + ) + if not _commit_or_rollback(db): + raise _PersistFailed("RecordRunMetrics did not persist") + return True + def _stash_tool_output(self, db, cmd: "StashToolOutput") -> bool: """Insert/upsert a large tool output's full text into `tool_outputs`. diff --git a/backend/app/claude_sdk_runner.py b/backend/app/claude_sdk_runner.py index 85000a773..1e61436b5 100644 --- a/backend/app/claude_sdk_runner.py +++ b/backend/app/claude_sdk_runner.py @@ -115,6 +115,7 @@ from app.sdk_emit import emit_unknown_enabled, unknown_event from app.tool_summaries import summarize_tool_input from app.tool_sources import normalize_tool_sources, sources_from_websearch_text +from app.usage_metrics import normalize_claude_usage log = logging.getLogger(__name__) @@ -1065,6 +1066,9 @@ def dispatch_sdk_message( "session_id": current_session_id, "cost_usd": sdk_msg.total_cost_usd, "usage": dict(sdk_msg.usage) if sdk_msg.usage else None, + "usage_metrics": normalize_claude_usage( + sdk_msg.usage, sdk_msg.model_usage, + ), "model_usage": ( dict(sdk_msg.model_usage) if sdk_msg.model_usage else None ), diff --git a/backend/app/codex_sdk_runner.py b/backend/app/codex_sdk_runner.py index fbd23d8eb..f8c630581 100644 --- a/backend/app/codex_sdk_runner.py +++ b/backend/app/codex_sdk_runner.py @@ -72,6 +72,7 @@ from app.codex_appserver import _extract_bash_command from app.providers import get_skill_path from app.runtime_types import RunnerResult +from app.usage_metrics import normalize_codex_usage from app.runner_registry import RunnerKind, registry from app.tool_sources import normalize_tool_sources @@ -1892,6 +1893,8 @@ async def run_codex_sdk_turn( current_session_id = session_id completed_turn: Any | None = None completed_message_phases: list[str | None] = [] + first_token_usage: Any | None = None + final_token_usage: Any | None = None process_group_id: int | None = None codex_context = sdk["AsyncCodex"](config=config) process_group_capture_stop: asyncio.Event | None = None @@ -2193,9 +2196,9 @@ def aborted_result() -> RunnerResult: continue if isinstance(payload, sdk["ThreadTokenUsageUpdatedNotification"]): - # Token usage is reported but currently not surfaced; the - # SDK already exposes it via the thread handle for any - # consumer that needs it. + if first_token_usage is None: + first_token_usage = payload.token_usage + final_token_usage = payload.token_usage continue if isinstance( @@ -2283,6 +2286,11 @@ def aborted_result() -> RunnerResult: "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: diff --git a/backend/app/database.py b/backend/app/database.py index 48d8c0855..1c366bf6c 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -724,6 +724,28 @@ def _slugify_for_source_dir(name: str) -> str: _add_runs.append( "ALTER TABLE chat_runs ADD COLUMN restart_nonce VARCHAR(128) NULL" ) + if "provider_session_id" not in chat_runs_cols: + _add_runs.append( + "ALTER TABLE chat_runs ADD COLUMN provider_session_id " + "VARCHAR(128) NULL" + ) + for column in ( + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "reasoning_output_tokens", + "total_tokens", + "model_context_window", + ): + if column not in chat_runs_cols: + _add_runs.append( + f"ALTER TABLE chat_runs ADD COLUMN {column} INTEGER NULL" + ) + if "usage_json" not in chat_runs_cols: + _add_runs.append( + "ALTER TABLE chat_runs ADD COLUMN usage_json JSON NULL" + ) if _add_runs: with eng.connect() as conn: for stmt in _add_runs: diff --git a/backend/app/main.py b/backend/app/main.py index 939f34bc3..ba898a83a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -529,38 +529,45 @@ async def _stalled_live_loop(): _stalled_live_task = _asyncio.create_task(_stalled_live_loop()) # Durable-continuation sweep (design §2.4): handles provider-limit resets - # and exact runs parked by a planned restart. It runs immediately on boot, - # then after a turn finishes or at the 60s fallback cadence. This is - # event-driven in the common path (one indexed query per completed turn), - # with no per-chat worker or short polling loop. + # and exact runs parked by a planned restart. The first pass is awaited + # BEFORE lifespan yields, so a reconnecting client cannot win the startup + # race and turn an opted-in restart into manual recovery. Later passes run + # after a turn finishes or at the 60s fallback cadence. This is event- + # driven in the common path (one indexed query per completed turn), with + # no per-chat worker or short polling loop. + from app.broadcast import get_system_broadcast as _system_broadcast + _reset_park_events = _system_broadcast().subscribe() + + async def _sweep_reset_parks_once(): + try: + _rp_db = _SweepSession() + try: + await sweep_reset_parks(_rp_db) + finally: + _rp_db.close() + except _asyncio.CancelledError: + raise + except Exception as _exc: + _log.error("reset-park sweep failed: %s", _exc, exc_info=True) + + await _sweep_reset_parks_once() + async def _reset_park_loop(): - from app.broadcast import get_system_broadcast as _system_broadcast - _events = _system_broadcast().subscribe() try: while True: - try: - _rp_db = _SweepSession() - try: - await sweep_reset_parks(_rp_db) - finally: - _rp_db.close() - except _asyncio.CancelledError: - raise - except Exception as _exc: - _log.error("reset-park sweep failed: %s", _exc, exc_info=True) - try: # One absolute fallback window. Unrelated system events must not # keep resetting a per-get timer and starve a future limit reset. async with _asyncio.timeout(60): while True: - _event = await _events.get() + _event = await _reset_park_events.get() if _event and _event.get("type") == "chat_run_finished": break except _asyncio.TimeoutError: pass + await _sweep_reset_parks_once() finally: - _system_broadcast().unsubscribe(_events) + _system_broadcast().unsubscribe(_reset_park_events) _reset_park_task = _asyncio.create_task(_reset_park_loop()) diff --git a/backend/app/models.py b/backend/app/models.py index 091b4314b..50064497f 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -250,9 +250,24 @@ class ChatRun(Base): initiated_by_app_id = Column( Integer, ForeignKey("apps.id"), nullable=True, default=None ) - # Reserved for per-run cost attribution (Capability B). No code path writes - # this yet, so reads are NULL until the Step-3b follow-up wires a producer. + # Per-run provider cost. Rows created before usage telemetry remain NULL. cost_usd = Column(Float, nullable=True, default=None) + # Provider session/thread id active for this run. This lets diagnostics + # distinguish a fresh provider context from a resumed one without relying on + # Chat.session_id, which is only the latest pointer. + provider_session_id = Column(String(128), nullable=True, default=None) + # Provider-neutral per-turn totals. `input_tokens` means total context + # processed for the turn, including cached input; the cache columns split + # that total where the provider exposes the distinction. Raw/provider- + # specific cumulative detail stays in usage_json for forward compatibility. + input_tokens = Column(Integer, nullable=True, default=None) + output_tokens = Column(Integer, nullable=True, default=None) + cache_read_input_tokens = Column(Integer, nullable=True, default=None) + cache_creation_input_tokens = Column(Integer, nullable=True, default=None) + reasoning_output_tokens = Column(Integer, nullable=True, default=None) + total_tokens = Column(Integer, nullable=True, default=None) + model_context_window = Column(Integer, nullable=True, default=None) + usage_json = Column(JSON, nullable=True, default=None) started_at = Column(DateTime, default=lambda: datetime.now(UTC)) ended_at = Column(DateTime, nullable=True, default=None) # Provider rate/usage-limit parking (design §2.4). When a turn dies on a @@ -674,6 +689,25 @@ class AppActivityState(Base): unseen = Column(Boolean, nullable=False, default=True, server_default=true()) +class AppPreviewState(Base): + """Durable acknowledgement of the exact app build opened from its chat CTA. + + This is separate from ``apps`` so acknowledging a preview never advances + ``App.updated_at`` — the executable-bundle version the acknowledgement is + meant to record. ``seen_as_final`` distinguishes opening a live preview from + opening the settled result: finishing the turn may surface the same build one + last time even when no final source write was needed. + """ + + __tablename__ = "app_preview_state" + + app_id = Column(Integer, ForeignKey("apps.id"), primary_key=True) + seen_updated_at = Column(DateTime, nullable=False) + seen_as_final = Column( + Boolean, nullable=False, default=False, server_default=false() + ) + + class PushSubscription(Base): """Browser push subscription for Web Push delivery.""" diff --git a/backend/app/resource_access.py b/backend/app/resource_access.py index 43f66e2f0..9d6924edf 100644 --- a/backend/app/resource_access.py +++ b/backend/app/resource_access.py @@ -16,14 +16,18 @@ """ from fastapi import HTTPException -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, load_only from app import models from app.deps import Principal def get_active_chat_for_principal( - db: Session, chat_id: str, principal: Principal, + db: Session, + chat_id: str, + principal: Principal, + *, + load_fields: tuple | None = None, ) -> models.Chat: """Fetches an active chat the principal may DRIVE, else 404/403. @@ -44,7 +48,12 @@ def get_active_chat_for_principal( the owner sees, so an app can't probe existence of chats it can't reach); 403 when an app token targets a chat it doesn't own. """ - chat = get_active_chat_or_404(db, chat_id) + required_fields = ( + (models.Chat.id, models.Chat.created_by_app_id, *load_fields) + if load_fields + else None + ) + chat = get_active_chat_or_404(db, chat_id, load_fields=required_fields) if principal.scope == "chat_embed" and principal.chat_id != chat_id: raise HTTPException( status_code=403, @@ -61,7 +70,10 @@ def get_active_chat_for_principal( def get_active_chat_or_404( - db: Session, chat_id: str, + db: Session, + chat_id: str, + *, + load_fields: tuple | None = None, ) -> models.Chat: """Fetches a non-soft-deleted Chat by id, raising 404 otherwise. @@ -84,7 +96,13 @@ def get_active_chat_or_404( Raises: HTTPException: 404 when no row matches OR the row is soft-deleted. """ - chat = db.query(models.Chat).filter( + query = db.query(models.Chat) + if load_fields: + # A narrow read must stay narrow. ``raiseload`` turns an accidental future + # field access into a test-visible failure rather than silently decoding a + # multi-megabyte transcript and erasing the endpoint's reason to exist. + query = query.options(load_only(*load_fields, raiseload=True)) + chat = query.filter( models.Chat.id == chat_id, models.Chat.deleted_at.is_(None), ).first() diff --git a/backend/app/routes/apps.py b/backend/app/routes/apps.py index f829a9688..94678a514 100644 --- a/backend/app/routes/apps.py +++ b/backend/app/routes/apps.py @@ -23,7 +23,7 @@ from sqlalchemy.orm import Session from app import ( - activity, app_activity, app_git, app_jobs, fs_locks, icon_cache, + activity, app_activity, app_git, app_jobs, app_preview, fs_locks, icon_cache, legacy_platform_apps, models, providers, schemas, source_dirs, theme, @@ -416,6 +416,9 @@ async def _hard_delete_app(db: Session, app: models.App) -> None: db.query(models.AppActivityState).filter( models.AppActivityState.app_id == deleted_app_id, ).delete(synchronize_session=False) + db.query(models.AppPreviewState).filter( + models.AppPreviewState.app_id == deleted_app_id, + ).delete(synchronize_session=False) db.delete(app) db.commit() get_system_broadcast().publish( @@ -702,7 +705,9 @@ async def list_apps( ) .all() ) - return app_activity.annotate_apps(db, apps) + return app_preview.annotate_apps( + db, app_activity.annotate_apps(db, apps) + ) @router.get("/schedules", response_model=list[schemas.AppScheduleOut]) @@ -1837,7 +1842,9 @@ def get_app( ): """Returns a single mini-app by ID (404 for a tombstoned one).""" app = live_app_or_404(db, app_id) - return app_activity.annotate_apps(db, [app])[0] + return app_preview.annotate_apps( + db, app_activity.annotate_apps(db, [app]) + )[0] class AppActivitySeenRequest(BaseModel): @@ -1862,6 +1869,43 @@ def mark_app_activity_seen( return Response(status_code=204) +class AppPreviewSeenRequest(BaseModel): + updated_at: datetime + final: bool = False + + +@router.post( + "/{app_id}/preview/seen", + status_code=204, + dependencies=[Depends(reject_cross_site)], +) +def mark_app_preview_seen( + app_id: int, + body: AppPreviewSeenRequest, + db: Session = Depends(get_db), + _: models.Owner = Depends(get_current_owner), +): + """Acknowledge the exact app build opened from its owning chat. + + The client sends the version it rendered, not merely the app id. If a newer + compile races this request, the older acknowledgement remains older and the + new build's CTA stays visible. + """ + app = live_app_or_404(db, app_id) + observed = app_preview.naive_utc(body.updated_at) + current = app_preview.naive_utc(app.updated_at) + if observed > current: + raise HTTPException( + status_code=409, + detail="Cannot acknowledge a preview newer than the installed app.", + ) + app_preview.mark_seen( + db, app_id, observed, seen_as_final=body.final, + ) + db.commit() + return Response(status_code=204) + + @router.patch( "/{app_id}", response_model=schemas.AppOut, @@ -2633,6 +2677,10 @@ def get_app_job_context( choices = resolve_background_agents(get_settings().data_dir, {}) return { "app_id": app_id, + # The supervisor binds the scheduled script to this exact app before + # granting its token and filesystem contract. This is non-secret durable + # identity, not owner configuration. + "source_dir": app.source_dir, "primary": choices.get("primary"), "fallback": choices.get("fallback"), # This is the same normalized, non-secret receipt the owner reviewed. diff --git a/backend/app/routes/chats.py b/backend/app/routes/chats.py index bb0afcf12..42362f3d2 100644 --- a/backend/app/routes/chats.py +++ b/backend/app/routes/chats.py @@ -292,9 +292,11 @@ def _chat_detail_response( limit: int = 20, before: int | None = None, expose_session: bool = True, + compact: bool = False, ) -> dict: """Canonical paginated chat payload shared by create and detail reads.""" from app.chat_transcript import ( + compact_messages_for_detail, historical_tool_output_ids, materialized_messages, project_messages_for_detail, @@ -338,6 +340,12 @@ def _chat_detail_response( fetchable_tool_output_ids=fetchable_tool_ids, live_message=live_message, ) + if compact: + page = compact_messages_for_detail( + page, + message_offset=start, + live_message=live_message, + ) provider = chat.provider or "claude" pending_question = questions.get(chat.id) @@ -1073,6 +1081,7 @@ def get_chat( chat_id: str, limit: int = 20, before: int | None = None, + compact: bool = False, principal: Principal = Depends(get_owner_or_chat_embed_principal), db: Session = Depends(get_db), ): @@ -1096,12 +1105,118 @@ def get_chat( db=db, limit=limit, before=before, + compact=compact, # Provider thread ids are backend continuity state, not part of the # embedded participant surface. expose_session=principal.scope != "chat_embed", ) +@router.get("/{chat_id}/runtime") +def get_chat_runtime( + chat_id: str, + principal: Principal = Depends(get_owner_or_chat_embed_principal), + db: Session = Depends(get_db), +): + """Return only the mutable runtime fields used by mounted chat controls.""" + if principal.scope == "app": + raise HTTPException(status_code=403, detail="App token is not valid here.") + require_chat_embed_operation(principal, "chat:read") + chat = get_active_chat_for_principal( + db, + chat_id, + principal, + load_fields=(models.Chat.pending_messages,), + ) + pending_question = questions.get(chat.id) + return { + "running": is_chat_running(chat.id), + "pending_messages": list(chat.pending_messages or []), + "pending_question_id": ( + pending_question.question_id if pending_question is not None else None + ), + } + + +@router.get("/{chat_id}/activity-detail") +def get_chat_activity_detail( + chat_id: str, + message_index: int = Query(ge=0), + start: int = Query(ge=0), + end: int = Query(gt=0), + principal: Principal = Depends(get_owner_or_chat_embed_principal), + db: Session = Depends(get_db), +): + """Return one deliberately expanded historical activity range. + + The ordinary transcript read projects multi-step activity into collapsed + metadata. This endpoint resolves that metadata back to the exact stored + blocks only after the owner opens the line. It never rewrites the transcript. + """ + from app.chat_transcript import ( + MAX_ACTIVITY_DETAIL_BLOCKS, + historical_tool_output_ids, + materialized_messages, + project_messages_for_detail, + ) + + if principal.scope == "app": + raise HTTPException(status_code=403, detail="App token is not valid here.") + require_chat_embed_operation(principal, "chat:read") + if end <= start or end - start > MAX_ACTIVITY_DETAIL_BLOCKS: + raise HTTPException(status_code=422, detail="Invalid activity range.") + + chat = get_active_chat_for_principal(db, chat_id, principal) + messages = materialized_messages(chat) + if message_index >= len(messages): + raise HTTPException(status_code=404, detail="Activity message not found.") + message = messages[message_index] + blocks = message.get("blocks") if isinstance(message, dict) else None + if not isinstance(blocks, list) or end > len(blocks): + raise HTTPException(status_code=404, detail="Activity range not found.") + + selected = [ + (raw_index, block) + for raw_index, block in enumerate(blocks[start:end], start=start) + if isinstance(block, dict) + and block.get("type") in {"tool", "thinking"} + and not ( + block.get("type") == "tool" + and block.get("tool") in {"AskUserQuestion", "request_user_input"} + and any( + isinstance(candidate, dict) and candidate.get("type") == "question" + for candidate in blocks + ) + ) + ] + detail_message = { + "role": "assistant", + "blocks": [block for _, block in selected], + } + candidate_tool_ids = historical_tool_output_ids([detail_message]) + fetchable_tool_ids = ( + { + row[0] + for row in db.query(models.ToolOutput.tool_use_id).filter( + models.ToolOutput.chat_id == chat.id, + models.ToolOutput.tool_use_id.in_(candidate_tool_ids), + ).all() + } + if candidate_tool_ids + else set() + ) + projected = project_messages_for_detail( + [detail_message], + fetchable_tool_output_ids=fetchable_tool_ids, + )[0]["blocks"] + return { + "entries": [ + {"item": block, "idx": raw_index} + for (raw_index, _), block in zip(selected, projected, strict=True) + ], + } + + @router.get("/{chat_id}/tool-output/{tool_use_id}", response_class=PlainTextResponse) def get_tool_output_by_id( chat_id: str, @@ -1323,6 +1438,71 @@ def get_chat_agent_context( } +@router.get("/{chat_id}/usage") +def get_chat_usage( + chat_id: str, + _: models.Owner = Depends(get_current_owner), + db: Session = Depends(get_db), +): + """Return provider-neutral token accounting for every durable chat run. + + Historical rows created before usage capture remain visible with null + counters, so callers can distinguish zero usage from missing coverage. + This owner-only diagnostic is deliberately independent of the transcript: + benchmark tooling can read it without parsing user-visible messages. + """ + get_active_chat_or_404(db, chat_id) + runs = ( + db.query(models.ChatRun) + .filter(models.ChatRun.chat_id == chat_id) + .order_by(models.ChatRun.started_at.asc(), models.ChatRun.id.asc()) + .all() + ) + count_fields = ( + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "reasoning_output_tokens", + "total_tokens", + ) + totals = {} + for field in count_fields: + values = [ + int(getattr(run, field)) + for run in runs + if getattr(run, field) is not None + ] + totals[field] = sum(values) if values else None + costs = [float(run.cost_usd) for run in runs if run.cost_usd is not None] + totals["cost_usd"] = sum(costs) if costs else None + + return { + "chat_id": chat_id, + "coverage": { + "runs": len(runs), + "runs_with_usage": sum(run.usage_json is not None for run in runs), + "runs_with_cost": len(costs), + }, + "totals": totals, + "runs": [ + { + "id": run.id, + "status": run.status, + "provider": run.provider, + "provider_session_id": run.provider_session_id, + "started_at": run.started_at, + "ended_at": run.ended_at, + "cost_usd": run.cost_usd, + **{field: getattr(run, field) for field in count_fields}, + "model_context_window": run.model_context_window, + "usage": run.usage_json, + } + for run in runs + ], + } + + @router.delete( "/{chat_id}", status_code=204, dependencies=[Depends(reject_cross_site)], ) diff --git a/backend/app/runtime_types.py b/backend/app/runtime_types.py index 773464c15..1ffa50379 100644 --- a/backend/app/runtime_types.py +++ b/backend/app/runtime_types.py @@ -12,6 +12,7 @@ class RunnerResult(TypedDict): cost_usd: float | None error: str | None usage: NotRequired[dict | None] + usage_metrics: NotRequired[dict | None] terminal_status: NotRequired[str | None] final_message_phase: NotRequired[str | None] diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 0d1ff6f83..a4f5f4308 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -113,6 +113,11 @@ class AppOut(BaseModel): # opened. The shell renders the same quiet activity dot used for chats. has_unseen_activity: bool = False unseen_activity_version: int | None = None + # Exact executable build last opened from the owning chat's CTA. Opening a + # live preview and opening the settled result are separate acknowledgements: + # the same build may surface once more when its agent turn finishes. + preview_seen_updated_at: datetime | None = None + preview_seen_final: bool = False cross_app_access: ShareLevel = "none" share_with_apps: ShareLevel = "none" offline_capable: bool = False diff --git a/backend/app/usage_metrics.py b/backend/app/usage_metrics.py new file mode 100644 index 000000000..7b21f8279 --- /dev/null +++ b/backend/app/usage_metrics.py @@ -0,0 +1,173 @@ +"""Provider-neutral per-turn token-usage normalization. + +Claude reports one aggregate usage dict on its terminal ResultMessage. Codex +reports a sequence of ThreadTokenUsage updates containing: + +- ``last``: the latest model call; +- ``total``: cumulative usage for the provider thread. + +For Codex, the first update implies the pre-turn baseline +(``first.total - first.last``). Subtracting that baseline from the final total +produces the sum of every model call in this Möbius turn — the quantity needed +to compare harness context efficiency. +""" + +from __future__ import annotations + +from typing import Any + + +def _count(value: Any) -> int: + """Return a non-negative integer counter; unknown SDK values become zero.""" + if isinstance(value, bool): + return 0 + try: + return max(0, int(value or 0)) + except (TypeError, ValueError): + return 0 + + +def _plain(value: Any) -> Any: + """Convert generated SDK models into JSON-safe plain values.""" + if value is None: + return None + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", by_alias=True) + if isinstance(value, dict): + return {str(k): _plain(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_plain(v) for v in value] + if isinstance(value, (str, int, float, bool)): + return value + return str(value) + + +def normalize_claude_usage( + usage: dict[str, Any] | None, + model_usage: dict[str, Any] | None = None, +) -> dict | None: + """Normalize Claude's terminal turn aggregate. + + Anthropic reports uncached input, cache creation, and cache reads as separate + counters. ``input_tokens`` below intentionally adds all three so it measures + total context processed/re-fed for the turn, matching the harness-efficiency + quantity Codex exposes. + """ + if not usage: + return None + uncached = _count(usage.get("input_tokens")) + cache_write = _count(usage.get("cache_creation_input_tokens")) + cache_read = _count(usage.get("cache_read_input_tokens")) + output = _count(usage.get("output_tokens")) + input_total = uncached + cache_write + cache_read + context_windows = [ + _count(details.get("contextWindow")) + for details in (model_usage or {}).values() + if isinstance(details, dict) + ] + return { + "provider": "claude", + "scope": "turn", + "calculation": "result_aggregate", + "input_tokens": input_total, + "uncached_input_tokens": uncached, + "output_tokens": output, + "cache_read_input_tokens": cache_read, + "cache_creation_input_tokens": cache_write, + "reasoning_output_tokens": _count(usage.get("reasoning_tokens")), + "total_tokens": input_total + output, + "model_context_window": max(context_windows, default=0) or None, + "provider_usage": _plain(usage), + "provider_model_usage": _plain(model_usage), + } + + +_CODEX_FIELDS = ( + "input_tokens", + "cached_input_tokens", + "output_tokens", + "reasoning_output_tokens", + "total_tokens", +) + + +def _codex_breakdown(value: Any) -> dict[str, int]: + if value is None: + return {field: 0 for field in _CODEX_FIELDS} + def read(field: str) -> Any: + if not isinstance(value, dict): + return getattr(value, field, None) + camel = field.split("_")[0] + "".join( + part.title() for part in field.split("_")[1:] + ) + return value.get(field, value.get(camel)) + return { + field: _count(read(field)) + for field in _CODEX_FIELDS + } + + +def _member(value: Any, field: str) -> Any: + if isinstance(value, dict): + camel = field.split("_")[0] + "".join( + part.title() for part in field.split("_")[1:] + ) + return value.get(field, value.get(camel)) + return getattr(value, field, None) + + +def _subtract_counts( + current: dict[str, int], + baseline: dict[str, int], +) -> dict[str, int]: + return { + field: max(0, current[field] - baseline[field]) + for field in _CODEX_FIELDS + } + + +def normalize_codex_usage( + first_usage: Any | None, + final_usage: Any | None, +) -> dict | None: + """Derive one Möbius-turn aggregate from Codex thread usage updates.""" + if final_usage is None: + return None + first_usage = first_usage or final_usage + first_total = _codex_breakdown(_member(first_usage, "total")) + first_last = _codex_breakdown(_member(first_usage, "last")) + baseline = _subtract_counts(first_total, first_last) + final_total = _codex_breakdown(_member(final_usage, "total")) + + # A cumulative provider counter should never go backwards. If an SDK/server + # reset makes it do so, the only honest bounded fallback is the latest call; + # retain the calculation label so benchmark consumers can exclude it. + if any(final_total[field] < baseline[field] for field in _CODEX_FIELDS): + turn = _codex_breakdown(_member(final_usage, "last")) + calculation = "last_call_fallback" + else: + turn = _subtract_counts(final_total, baseline) + calculation = "thread_delta" + + input_total = turn["input_tokens"] + cached = min(turn["cached_input_tokens"], input_total) + return { + "provider": "codex", + "scope": "turn", + "calculation": calculation, + "input_tokens": input_total, + "uncached_input_tokens": max(0, input_total - cached), + "output_tokens": turn["output_tokens"], + "cache_read_input_tokens": cached, + "cache_creation_input_tokens": 0, + "reasoning_output_tokens": turn["reasoning_output_tokens"], + "total_tokens": turn["total_tokens"], + "model_context_window": _count( + _member(final_usage, "model_context_window") + ) or None, + "provider_thread_total": final_total, + "provider_usage": { + "first": _plain(first_usage), + "final": _plain(final_usage), + }, + } diff --git a/backend/scripts/app-job-runner.py b/backend/scripts/app-job-runner.py index c5490fe80..bdfa6c623 100755 --- a/backend/scripts/app-job-runner.py +++ b/backend/scripts/app-job-runner.py @@ -97,6 +97,24 @@ def _job_context(app_id: int, token: str) -> dict | None: return None +def _job_matches_context(resolved: Path, context: dict) -> bool: + """Bind a scheduled script to the exact app whose authority it receives. + + Missing, malformed, or mismatched identity fails closed before the app token + reaches a child process. During a platform-update window, an older backend + may omit ``source_dir``; skipping that run is safer than retaining a + permanent compatibility path around this capability boundary. + """ + source_dir = context.get("source_dir") + if not isinstance(source_dir, str) or not source_dir: + return False + try: + expected = Path(source_dir).resolve(strict=True) + except (OSError, RuntimeError): + return False + return expected == resolved.parent + + def _mint_app_token(app_id: int) -> str | None: """Exchange the owner service credential for one short-lived app token.""" try: @@ -281,6 +299,9 @@ def run() -> int: if context is None: _log(app_id, "failed: job-context fetch") return 4 + if not _job_matches_context(resolved, context): + _log(app_id, f"rejected: job does not belong to app: {resolved}") + return 4 command = _sandboxed_command(app_id, resolved, context) if command is None: _log(app_id, "failed: sandbox unavailable for background agent") diff --git a/backend/tests/test_app_jobs.py b/backend/tests/test_app_jobs.py index e2eafe392..523778e62 100644 --- a/backend/tests/test_app_jobs.py +++ b/backend/tests/test_app_jobs.py @@ -155,7 +155,11 @@ def test_wrapper_runs_job_only_after_live_check(tmp_path, monkeypatch): runner, "_app_is_live", lambda app_id, token=None: True, ) monkeypatch.setattr(runner.os, "getsid", lambda _pid: os.getpid()) - monkeypatch.setattr(runner, "_job_context", lambda app_id, token: {}) + monkeypatch.setattr( + runner, + "_job_context", + lambda app_id, token: {"source_dir": str(source)}, + ) popen = types.SimpleNamespace(wait=lambda: 0) calls = [] monkeypatch.setattr( @@ -176,6 +180,69 @@ def test_wrapper_runs_job_only_after_live_check(tmp_path, monkeypatch): assert "AGENT_TOKEN" not in child_env +def test_wrapper_rejects_a_job_from_another_live_app(tmp_path, monkeypatch): + runner = _load_runner() + data_dir = tmp_path / "data" + memory_source = data_dir / "apps" / "memory" + reflection_source = data_dir / "apps" / "reflection" + memory_source.mkdir(parents=True) + reflection_source.mkdir() + job = memory_source / "fetch.sh" + job.write_text("#!/bin/sh\nexit 0\n") + monkeypatch.setattr(runner, "DATA_DIR", data_dir) + monkeypatch.setattr(runner, "_mint_app_token", lambda app_id: "app-token") + monkeypatch.setattr( + runner, "_app_is_live", lambda app_id, token=None: True, + ) + monkeypatch.setattr( + runner, + "_job_context", + lambda app_id, token: {"source_dir": str(reflection_source)}, + ) + monkeypatch.setattr(runner.os, "getsid", lambda _pid: os.getpid()) + calls = [] + monkeypatch.setattr( + runner.subprocess, + "Popen", + lambda *args, **kwargs: calls.append((args, kwargs)), + ) + monkeypatch.setattr(runner.sys, "argv", [ + "app-job-runner.py", "56", str(job), + ]) + + assert runner.run() == 4 + assert calls == [] + + +def test_wrapper_rejects_job_context_without_exact_app_identity( + tmp_path, monkeypatch, +): + runner = _load_runner() + source = tmp_path / "data" / "apps" / "memory" + source.mkdir(parents=True) + job = source / "fetch.sh" + job.write_text("#!/bin/sh\nexit 0\n") + monkeypatch.setattr(runner, "DATA_DIR", tmp_path / "data") + monkeypatch.setattr(runner, "_mint_app_token", lambda app_id: "app-token") + monkeypatch.setattr( + runner, "_app_is_live", lambda app_id, token=None: True, + ) + monkeypatch.setattr(runner, "_job_context", lambda app_id, token: {}) + monkeypatch.setattr(runner.os, "getsid", lambda _pid: os.getpid()) + calls = [] + monkeypatch.setattr( + runner.subprocess, + "Popen", + lambda *args, **kwargs: calls.append((args, kwargs)), + ) + monkeypatch.setattr(runner.sys, "argv", [ + "app-job-runner.py", "56", str(job), + ]) + + assert runner.run() == 4 + assert calls == [] + + def test_background_agent_command_masks_platform_data_and_mounts_declared_scope( tmp_path, monkeypatch, ): @@ -321,6 +388,7 @@ def test_job_context_is_nonsecret_and_self_scoped(client, owner_token, db): assert response.status_code == 200, response.text body = response.json() assert body["app_id"] == own.id + assert body["source_dir"] == own.source_dir serialized = json.dumps(body).lower() assert "token" not in serialized assert "credential" not in serialized diff --git a/backend/tests/test_app_preview.py b/backend/tests/test_app_preview.py new file mode 100644 index 000000000..670ab60ba --- /dev/null +++ b/backend/tests/test_app_preview.py @@ -0,0 +1,91 @@ +"""Owning-chat app CTAs acknowledge exact preview builds durably.""" + +from datetime import timedelta + +from app import models + + +def _app(db): + app = models.App( + name="Atlas", description="", chat_id="chat-a", + jsx_source="export default function App(){}", + compiled_path="/tmp/app.js", + ) + db.add(app) + db.commit() + db.refresh(app) + return app + + +def _listed(client, auth, app_id): + response = client.get("/api/apps/", headers=auth) + assert response.status_code == 200, response.text + return next(row for row in response.json() if row["id"] == app_id) + + +def test_preview_then_final_acknowledgement_is_durable(client, auth, db): + app = _app(db) + row = _listed(client, auth, app.id) + assert row["preview_seen_updated_at"] is None + assert row["preview_seen_final"] is False + + preview = client.post( + f"/api/apps/{app.id}/preview/seen", + headers=auth, + json={"updated_at": row["updated_at"], "final": False}, + ) + assert preview.status_code == 204, preview.text + row = _listed(client, auth, app.id) + assert row["preview_seen_updated_at"] == row["updated_at"] + assert row["preview_seen_final"] is False + + final = client.post( + f"/api/apps/{app.id}/preview/seen", + headers=auth, + json={"updated_at": row["updated_at"], "final": True}, + ) + assert final.status_code == 204, final.text + row = _listed(client, auth, app.id) + assert row["preview_seen_updated_at"] == row["updated_at"] + assert row["preview_seen_final"] is True + + +def test_stale_open_never_hides_a_newer_build(client, auth, db): + app = _app(db) + old_version = app.updated_at + + app.updated_at = old_version + timedelta(seconds=1) + db.commit() + db.refresh(app) + new_version = app.updated_at + + current = client.post( + f"/api/apps/{app.id}/preview/seen", + headers=auth, + json={"updated_at": new_version.isoformat(), "final": False}, + ) + assert current.status_code == 204, current.text + + # A delayed click from another pane/device cannot move the acknowledgement + # back or incorrectly promote the current build to its final phase. + stale = client.post( + f"/api/apps/{app.id}/preview/seen", + headers=auth, + json={"updated_at": old_version.isoformat(), "final": True}, + ) + assert stale.status_code == 204, stale.text + row = _listed(client, auth, app.id) + assert row["preview_seen_updated_at"] == row["updated_at"] + assert row["preview_seen_final"] is False + + +def test_future_preview_version_is_rejected(client, auth, db): + app = _app(db) + future = app.updated_at + timedelta(days=1) + response = client.post( + f"/api/apps/{app.id}/preview/seen", + headers=auth, + json={"updated_at": future.isoformat(), "final": True}, + ) + assert response.status_code == 409, response.text + assert db.get(models.AppPreviewState, app.id) is None diff --git a/backend/tests/test_chat_runs.py b/backend/tests/test_chat_runs.py index f2459fbed..1ebf14b07 100644 --- a/backend/tests/test_chat_runs.py +++ b/backend/tests/test_chat_runs.py @@ -14,7 +14,7 @@ from app import models from app.chat_writer import ( AppendPending, Barrier, ClearRunStatus, PromotePending, StartTurn, - alloc_run_token, get_writer, + RecordRunMetrics, alloc_run_token, get_writer, ) from app.database import SessionLocal @@ -127,6 +127,49 @@ def test_clear_run_status_preserves_failed_outcome(): assert _run_status("r2-failed") is None +def test_record_run_metrics_updates_exact_run_without_touching_transcript(): + _seed_chat("r-metrics", messages=[{ + "role": "user", "content": "keep me", "ts": 1, + }]) + _seed_run("rt-metrics", "r-metrics") + + get_writer().submit(RecordRunMetrics( + chat_id="r-metrics", + run_token="rt-metrics", + provider_session_id="provider-thread", + cost_usd=0.125, + usage={ + "provider": "codex", + "input_tokens": 900, + "output_tokens": 200, + "cache_read_input_tokens": 500, + "cache_creation_input_tokens": 0, + "reasoning_output_tokens": 100, + "total_tokens": 1_100, + "model_context_window": 200_000, + }, + )).result(timeout=5) + + db = SessionLocal() + try: + run = db.query(models.ChatRun).filter( + models.ChatRun.id == "rt-metrics", + ).one() + chat = db.query(models.Chat).filter(models.Chat.id == "r-metrics").one() + assert run.provider_session_id == "provider-thread" + assert run.cost_usd == 0.125 + assert run.input_tokens == 900 + assert run.output_tokens == 200 + assert run.cache_read_input_tokens == 500 + assert run.reasoning_output_tokens == 100 + assert run.total_tokens == 1_100 + assert run.model_context_window == 200_000 + assert run.usage_json["provider"] == "codex" + assert chat.messages == [{"role": "user", "content": "keep me", "ts": 1}] + finally: + db.close() + + # -- continuation handoff ------------------------------------------------- def test_promote_closes_prior_run_and_opens_the_continuation(): _seed_chat("r3") diff --git a/backend/tests/test_chat_transcript_compact.py b/backend/tests/test_chat_transcript_compact.py new file mode 100644 index 000000000..e99683770 --- /dev/null +++ b/backend/tests/test_chat_transcript_compact.py @@ -0,0 +1,342 @@ +"""Compact chat reads keep the transcript light without changing stored truth.""" + +from app.chat_transcript import compact_messages_for_detail +from sqlalchemy import event + + +def test_compacts_multi_step_activity_and_preserves_render_metadata(): + source = {"title": "Reference", "url": "https://example.com/reference"} + messages = [{ + "role": "assistant", + "content": "Answer", + "blocks": [ + {"type": "text", "content": "Before"}, + {"type": "thinking", "thinking_id": "thought-1", "duration_ms": 1200}, + { + "type": "tool", + "tool": "WebSearch", + "tool_use_id": "tool-1", + "status": "done", + "input": {"query": "large private input"}, + "output": "large output", + "sources": [source], + "subagent": {"helper": {"status": "done"}}, + }, + { + "type": "tool", + "tool": "Bash", + "tool_use_id": "tool-2", + "status": "done", + "output": "more output", + "output_exit_code": 0, + }, + {"type": "text", "content": "After"}, + ], + }] + + compact = compact_messages_for_detail(messages, message_offset=40) + + assert compact is not messages + assert compact[0] is not messages[0] + assert "content" not in compact[0] + assert compact[0]["blocks"][0] == {"type": "text", "content": "Before"} + summary = compact[0]["blocks"][1] + assert summary == { + "type": "activity", + "activity_id": "40:1:4", + "message_index": 40, + "start": 1, + "end": 4, + "tool_count": 2, + "entries": [ + { + "item": { + "type": "thinking", + "thinking_id": "thought-1", + "duration_ms": 1200, + }, + "idx": 1, + }, + { + "item": { + "type": "tool", + "tool": "WebSearch", + "status": "done", + "tool_use_id": "tool-1", + "subagent": {"helper": {"status": "done"}}, + }, + "idx": 2, + }, + { + "item": { + "type": "tool", + "tool": "Bash", + "status": "done", + "tool_use_id": "tool-2", + "output_exit_code": 0, + }, + "idx": 3, + }, + ], + "sources": [source], + } + assert compact[0]["blocks"][2] == {"type": "text", "content": "After"} + assert messages[0]["blocks"][2]["output"] == "large output" + + +def test_repeated_activity_metadata_is_bounded_by_variety(): + blocks = [ + {"type": "thinking", "duration_ms": 100}, + *[ + { + "type": "tool", + "tool": "Bash", + "status": "done", + "output": f"step {index}", + } + for index in range(100) + ], + {"type": "thinking", "duration_ms": 200}, + {"type": "tool", "tool": "Edit", "status": "done"}, + ] + + compact = compact_messages_for_detail( + [{"role": "assistant", "blocks": blocks}], + message_offset=0, + ) + summary = compact[0]["blocks"][0] + + assert summary["tool_count"] == 101 + assert len(summary["entries"]) == 4 + assert summary["entries"][0]["item"]["duration_ms"] == 300 + assert [ + entry["item"].get("tool") + for entry in summary["entries"][1:] + ] == ["Bash", "Bash", "Edit"] + + +def test_long_activity_runs_are_split_into_fetchable_ranges(): + blocks = [ + { + "type": "tool", + "tool": "Bash", + "status": "done", + "output": f"step {index}", + } + for index in range(2001) + ] + + compact = compact_messages_for_detail( + [{"role": "assistant", "blocks": blocks}], + message_offset=4, + ) + + assert compact[0]["blocks"] == [ + { + **compact[0]["blocks"][0], + "activity_id": "4:0:2000", + "message_index": 4, + "start": 0, + "end": 2000, + "tool_count": 2000, + }, + blocks[2000], + ] + assert compact[0]["blocks"][0]["type"] == "activity" + assert compact[0]["blocks"][0]["end"] - compact[0]["blocks"][0]["start"] == 2000 + + +def test_single_activity_and_live_message_remain_self_contained(): + single = { + "role": "assistant", + "content": "One step", + "blocks": [ + {"type": "tool", "tool": "Read", "input": "/tmp/note.txt"}, + {"type": "text", "content": "Done"}, + ], + } + live = { + "role": "assistant", + "blocks": [ + {"type": "thinking", "content": "working"}, + {"type": "tool", "tool": "Bash", "output": "live"}, + ], + } + messages = [single, live] + + compact = compact_messages_for_detail( + messages, + message_offset=0, + live_message=live, + ) + + assert compact is messages + assert compact[0] is single + assert compact[1] is live + + +def test_image_reads_stay_distinctive_and_question_twins_are_not_rendered(): + messages = [{ + "role": "assistant", + "content": "Picked", + "blocks": [ + {"type": "thinking", "duration_ms": 10}, + {"type": "tool", "tool": "Read", "input": "/tmp/photo.webp"}, + {"type": "tool", "tool": "Bash", "output": "one"}, + {"type": "thinking", "duration_ms": 20}, + {"type": "tool", "tool": "request_user_input", "status": "done"}, + {"type": "question", "question_id": "q1", "questions": []}, + ], + }] + + compact = compact_messages_for_detail(messages, message_offset=7) + blocks = compact[0]["blocks"] + + assert blocks[0]["type"] == "thinking" + assert blocks[1] == messages[0]["blocks"][1] + assert blocks[2]["type"] == "activity" + assert blocks[2]["start"] == 2 + assert blocks[2]["end"] == 4 + assert blocks[3]["type"] == "question" + assert all( + block.get("tool") != "request_user_input" + for block in blocks + if isinstance(block, dict) + ) + + +def test_compact_route_defers_activity_detail_until_expansion(client, auth): + messages = [{ + "role": "assistant", + "content": "Complete answer", + "blocks": [ + {"type": "thinking", "content": "private trace", "duration_ms": 500}, + { + "type": "tool", + "tool": "Bash", + "tool_use_id": "tool-full", + "status": "done", + "input": "printf hello", + "output": "hello", + }, + {"type": "text", "content": "Complete answer"}, + ], + }] + created = client.post( + "/api/chats", + headers=auth, + json={"title": "Compact route", "messages": messages}, + ) + assert created.status_code == 200 + chat_id = created.json()["id"] + + compact = client.get( + f"/api/chats/{chat_id}?limit=20&compact=1", + headers=auth, + ) + assert compact.status_code == 200 + summary = compact.json()["messages"][0]["blocks"][0] + assert summary["type"] == "activity" + assert "content" not in summary["entries"][0]["item"] + assert "output" not in summary["entries"][1]["item"] + + detail = client.get( + f"/api/chats/{chat_id}/activity-detail" + "?message_index=0&start=0&end=2", + headers=auth, + ) + assert detail.status_code == 200 + entries = detail.json()["entries"] + assert entries[0]["item"]["content"] == "private trace" + assert entries[1]["item"]["output"] == "hello" + + +def test_activity_detail_queries_only_candidate_tool_sidecars( + client, + auth, + db, +): + messages = [{ + "role": "assistant", + "blocks": [ + {"type": "thinking", "content": "trace"}, + { + "type": "tool", + "tool": "Bash", + "tool_use_id": "tool-candidate", + "status": "done", + "output": "preview", + "output_truncated": True, + }, + ], + }] + created = client.post( + "/api/chats", + headers=auth, + json={"title": "Scoped sidecars", "messages": messages}, + ) + chat_id = created.json()["id"] + statements = [] + engine = db.get_bind() + + def capture_sql(_, __, statement, *args): + statements.append(statement.lower()) + + event.listen(engine, "before_cursor_execute", capture_sql) + try: + detail = client.get( + f"/api/chats/{chat_id}/activity-detail" + "?message_index=0&start=0&end=2", + headers=auth, + ) + finally: + event.remove(engine, "before_cursor_execute", capture_sql) + + assert detail.status_code == 200 + sidecar_select = next( + statement + for statement in statements + if "from tool_outputs" in statement + ) + assert "tool_outputs.chat_id =" in sidecar_select + assert "tool_outputs.tool_use_id in (" in sidecar_select + + +def test_runtime_route_does_not_select_transcript_json( + client, + auth, + db, + monkeypatch, +): + created = client.post( + "/api/chats", + headers=auth, + json={"title": "Runtime projection"}, + ) + chat_id = created.json()["id"] + monkeypatch.setattr("app.routes.chats.is_chat_running", lambda _: True) + statements = [] + engine = db.get_bind() + + def capture_sql(_, __, statement, *args): + statements.append(statement.lower()) + + event.listen(engine, "before_cursor_execute", capture_sql) + try: + runtime = client.get(f"/api/chats/{chat_id}/runtime", headers=auth) + finally: + event.remove(engine, "before_cursor_execute", capture_sql) + + assert runtime.status_code == 200 + assert runtime.json() == { + "running": True, + "pending_messages": [], + "pending_question_id": None, + } + chat_select = next( + statement + for statement in statements + if "from chats" in statement and "chats.pending_messages" in statement + ) + assert "chats.pending_messages" in chat_select + assert "chats.messages as" not in chat_select diff --git a/backend/tests/test_chats.py b/backend/tests/test_chats.py index fb81b0b56..f08382e72 100644 --- a/backend/tests/test_chats.py +++ b/backend/tests/test_chats.py @@ -1,6 +1,7 @@ """Chat route regression tests.""" import asyncio +from datetime import UTC, datetime import io from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 @@ -130,6 +131,63 @@ def test_agent_context_includes_evolving_chat_summary( assert payload["system_prompt_origin"] == "platform" +def test_chat_usage_reports_totals_and_historic_coverage( + client, auth, chat, db, +): + db.add_all([ + models.ChatRun( + id="historic-run", + chat_id=chat.id, + status="completed", + provider="claude", + started_at=datetime.now(UTC), + ), + models.ChatRun( + id="measured-run", + chat_id=chat.id, + status="completed", + provider="codex", + provider_session_id="thread-1", + cost_usd=0.125, + input_tokens=900, + output_tokens=200, + cache_read_input_tokens=500, + cache_creation_input_tokens=0, + reasoning_output_tokens=100, + total_tokens=1_100, + model_context_window=200_000, + usage_json={"provider": "codex", "calculation": "thread_delta"}, + started_at=datetime.now(UTC), + ), + ]) + db.commit() + + response = client.get(f"/api/chats/{chat.id}/usage", headers=auth) + + assert response.status_code == 200 + payload = response.json() + assert payload["coverage"] == { + "runs": 2, + "runs_with_usage": 1, + "runs_with_cost": 1, + } + assert payload["totals"] == { + "input_tokens": 900, + "output_tokens": 200, + "cache_read_input_tokens": 500, + "cache_creation_input_tokens": 0, + "reasoning_output_tokens": 100, + "total_tokens": 1_100, + "cost_usd": 0.125, + } + measured = next( + run for run in payload["runs"] if run["id"] == "measured-run" + ) + assert measured["provider_session_id"] == "thread-1" + assert measured["model_context_window"] == 200_000 + assert measured["usage"]["calculation"] == "thread_delta" + + def test_create_chat_rejects_cross_site_request(client, auth): cross = client.post( "/api/chats", diff --git a/backend/tests/test_claude_sdk_runner.py b/backend/tests/test_claude_sdk_runner.py index c874fc661..7b7fcae34 100644 --- a/backend/tests/test_claude_sdk_runner.py +++ b/backend/tests/test_claude_sdk_runner.py @@ -1260,6 +1260,21 @@ def test_dispatch_result_message_returns_terminal(): assert terminal["cost_usd"] == 0.05 assert terminal["session_id"] == "sess-1" assert terminal["usage"] == {"input_tokens": 100, "output_tokens": 200} + assert terminal["usage_metrics"] == { + "provider": "claude", + "scope": "turn", + "calculation": "result_aggregate", + "input_tokens": 100, + "uncached_input_tokens": 100, + "output_tokens": 200, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "reasoning_output_tokens": 0, + "total_tokens": 300, + "model_context_window": None, + "provider_usage": {"input_tokens": 100, "output_tokens": 200}, + "provider_model_usage": None, + } # ResultMessage also fires usage + stop_reason side-channels. types = [e["type"] for e in bus.events] assert "usage" in types diff --git a/backend/tests/test_codex_sdk_runner.py b/backend/tests/test_codex_sdk_runner.py index a216b0a40..0404459c4 100644 --- a/backend/tests/test_codex_sdk_runner.py +++ b/backend/tests/test_codex_sdk_runner.py @@ -676,6 +676,99 @@ async def thread_resume(self, *_args, **_kwargs): assert registry.get_handle("chat-1", RunnerKind.CODEX_SDK) is None +def test_run_codex_sdk_turn_reports_all_model_calls_in_turn(monkeypatch): + class TokenUsageUpdated: + def __init__(self, token_usage): + self.token_usage = token_usage + + class Usage: + def __init__(self, last, total): + self.last = SimpleNamespace(**last) + self.total = SimpleNamespace(**total) + 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, + } + + first = Usage( + last={ + "input_tokens": 200, "cached_input_tokens": 100, + "output_tokens": 100, "reasoning_output_tokens": 50, + "total_tokens": 300, + }, + total={ + "input_tokens": 1_000, "cached_input_tokens": 400, + "output_tokens": 100, "reasoning_output_tokens": 50, + "total_tokens": 1_100, + }, + ) + final = Usage( + last={ + "input_tokens": 300, "cached_input_tokens": 150, + "output_tokens": 100, "reasoning_output_tokens": 50, + "total_tokens": 400, + }, + total={ + "input_tokens": 1_700, "cached_input_tokens": 800, + "output_tokens": 200, "reasoning_output_tokens": 100, + "total_tokens": 1_900, + }, + ) + completed_turn = SimpleNamespace(id="turn-1", usage=None, error=None) + notifications = [ + SimpleNamespace( + method="thread/tokenUsage/updated", + payload=TokenUsageUpdated(first), + ), + SimpleNamespace( + method="thread/tokenUsage/updated", + payload=TokenUsageUpdated(final), + ), + SimpleNamespace( + method="turn/completed", + payload=_FakeTurnCompletedNotification(completed_turn), + ), + ] + thread = _FakeThread("thread-usage", _FakeTurnHandle(notifications)) + + 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["ThreadTokenUsageUpdatedNotification"] = TokenUsageUpdated + monkeypatch.setattr(codex_sdk_runner, "_sdk_imports", lambda: sdk) + + result = asyncio.run(codex_sdk_runner.run_codex_sdk_turn( + user_message="measure this turn", + session_id=None, + base_env={}, + cwd="/tmp", + chat_id="chat-usage", + bc=_FakeBroadcast(), + pending_questions={}, + db=None, + )) + + assert result["usage"]["total"]["total_tokens"] == 1_900 + assert result["usage_metrics"]["calculation"] == "thread_delta" + assert result["usage_metrics"]["input_tokens"] == 900 + assert result["usage_metrics"]["total_tokens"] == 1_100 + + def test_run_codex_sdk_turn_resume_validation_error_now_propagates(monkeypatch, caplog): # Inverse of the old subAgentActivity resume test. The SDK now models # subAgentActivity natively, so thread_resume no longer raises on that diff --git a/backend/tests/test_db_migrations.py b/backend/tests/test_db_migrations.py index 8853889b9..afb91932b 100644 --- a/backend/tests/test_db_migrations.py +++ b/backend/tests/test_db_migrations.py @@ -109,6 +109,17 @@ def test_run_migrations_adds_park_columns_to_existing_chat_runs(tmp_path): assert "parked_until" in cols assert "park_reason" in cols assert "restart_nonce" in cols + assert { + "provider_session_id", + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "reasoning_output_tokens", + "total_tokens", + "model_context_window", + "usage_json", + } <= cols def test_agent_lifecycle_width_migration_is_postgres_only_and_idempotent(): diff --git a/backend/tests/test_limit_park.py b/backend/tests/test_limit_park.py index 88eba8fa4..18c1bcd7c 100644 --- a/backend/tests/test_limit_park.py +++ b/backend/tests/test_limit_park.py @@ -14,9 +14,10 @@ an opted park retryable until its continuation starts, skips future parks, stands down while draining, and resolves deleted chats silently. - (e) Auto-resume is policy-controlled (off = notify only) and strictly - serial: one park per tick, none while any turn is live; the resumed turn - combines the preserved queue + a "continue" into one continuation. + (e) Auto-resume is policy-controlled (off = notify only). Provider-limit + retries are strictly serial, while an accepted planned restart resumes + the exact previously-live set together. Each resumed turn combines its + preserved queue + a "continue" into one continuation. (f) The parks are observable: /api/debug/status lists parked runs. (g) A planned restart reuses the same exact-run state with a due-now time; crashes, unanswered questions, and app-owned work stay manual. @@ -1204,6 +1205,39 @@ def test_sweep_starts_only_one_of_two_opted_chats(owner_token, monkeypatch): chat_mod.discard_starting(cid) +def test_sweep_restarts_every_opted_chat_in_the_accepted_batch( + owner_token, monkeypatch, +): + """A restart restores the exact set that was already concurrent.""" + del owner_token + nonce = "restart-nonce-batch" + monkeypatch.setattr( + "app.restart_ledger.authorized_restart_nonce", lambda: nonce, + ) + monkeypatch.setattr( + "app.push.notify_owner", lambda *args, **kwargs: "notif-id", + ) + scheduled = [] + monkeypatch.setattr( + chat_mod, "_schedule_continuation", lambda **kw: scheduled.append(kw), + ) + chat_ids = ("restart-batch-a", "restart-batch-b", "restart-batch-c") + for cid in chat_ids: + _due_park( + cid, f"rt-{cid}", auto_restart=True, + park_reason="restart", restart_nonce=nonce, + ) + + try: + assert set(_run_sweep()) == set(chat_ids) + assert {item["chat_id"] for item in scheduled} == set(chat_ids) + for cid in chat_ids: + assert _run_row(f"rt-{cid}")["status"] == "completed" + finally: + for cid in chat_ids: + chat_mod.discard_starting(cid) + + def test_auto_resume_spawn_failure_rolls_back_and_retries_once( owner_token, monkeypatch, ): diff --git a/backend/tests/test_routes_init_resilience.py b/backend/tests/test_routes_init_resilience.py index 5f71c16ed..59184de7e 100644 --- a/backend/tests/test_routes_init_resilience.py +++ b/backend/tests/test_routes_init_resilience.py @@ -12,8 +12,10 @@ collapse when a real route module is forced to fail. """ +import asyncio import importlib import sys +import threading import pytest from fastapi import APIRouter, FastAPI @@ -158,6 +160,47 @@ def boom(loop): assert body["boot_id"] +def test_lifespan_waits_for_initial_restart_resume_sweep(monkeypatch): + """The server must not accept a manual send before restart recovery claims.""" + from app import chat as chat_mod + from app.main import app as main_app + + sweep_entered = threading.Event() + release_sweep = threading.Event() + lifespan_ready = threading.Event() + boot_errors = [] + + async def held_sweep(db): + del db + sweep_entered.set() + await asyncio.to_thread(release_sweep.wait) + return [] + + monkeypatch.setattr(chat_mod, "sweep_reset_parks", held_sweep) + + def boot_app(): + try: + with TestClient(main_app): + lifespan_ready.set() + except BaseException as exc: # surface a lifespan-thread failure below + boot_errors.append(exc) + + thread = threading.Thread(target=boot_app, daemon=True) + thread.start() + try: + assert sweep_entered.wait(timeout=20) + # The old fire-and-forget startup reached the usable server while this + # sweep was still blocked. The fixed lifecycle awaits it before yielding. + assert not lifespan_ready.wait(timeout=1.0) + finally: + release_sweep.set() + + thread.join(timeout=30) + assert not thread.is_alive() + assert boot_errors == [] + assert lifespan_ready.is_set() + + def test_lifespan_does_not_shadow_module_session_factory(): """A late local import must not break earlier startup migrations. diff --git a/backend/tests/test_runtime_libs.py b/backend/tests/test_runtime_libs.py index 1f146db1e..f34ae8f5f 100644 --- a/backend/tests/test_runtime_libs.py +++ b/backend/tests/test_runtime_libs.py @@ -67,6 +67,9 @@ def test_supported_runtime_packages_are_production_dependencies(): def test_compile_command_bundles_the_complete_runtime_graph(): command = esbuild_command("entry.jsx", "app.js") assert "--bundle" in command + assert '--define:process.env.NODE_ENV="production"' in command + assert "--minify" in command + assert "--keep-names" in command assert not any(arg.startswith("--external:") for arg in command) assert f"--banner:js={COMPILED_RUNTIME_BANNER}" in command assert f"--inject:{runtime_inject_path()}" in command @@ -79,6 +82,48 @@ def test_compile_command_bundles_the_complete_runtime_graph(): assert mobius_runtime_path().is_file() +def test_compile_command_selects_production_react_and_keeps_one_module(tmp_path): + """The size win must come from the real production graph, not externals.""" + entry = tmp_path / "entry.jsx" + output = tmp_path / "app.js" + metafile = tmp_path / "app-meta.json" + entry.write_text( + """import { useState } from 'react' + +export default function NamedFixture() { + const [value] = useState('ready') + return
{value}
+} +""" + ) + + completed = subprocess.run( + esbuild_command(entry, output, metafile=metafile), + capture_output=True, + check=False, + env=esbuild_environment(), + text=True, + timeout=ESBUILD_TIMEOUT_SECS, + ) + assert completed.returncode == 0, completed.stderr + + metadata = json.loads(metafile.read_text()) + inputs = set(metadata["inputs"]) + react_inputs = {name for name in inputs if "/react" in name} + assert react_inputs + assert not any(".development.js" in name for name in react_inputs) + assert any(".production.js" in name for name in react_inputs) + + entry_outputs = [ + details for details in metadata["outputs"].values() + if details.get("entryPoint") + ] + assert len(entry_outputs) == 1 + assert entry_outputs[0].get("imports") == [] + assert output.stat().st_size < 400_000 + assert "NamedFixture" in output.read_text() + + def test_app_local_or_transitive_react_cannot_shadow_platform_runtime(tmp_path): """App-local dependencies must not create a second React dispatcher.""" local_react = tmp_path / "node_modules" / "react" diff --git a/backend/tests/test_usage_metrics.py b/backend/tests/test_usage_metrics.py new file mode 100644 index 000000000..e6c6c1bcc --- /dev/null +++ b/backend/tests/test_usage_metrics.py @@ -0,0 +1,142 @@ +"""Provider-neutral token accounting regression tests.""" + +from types import SimpleNamespace + +from app.usage_metrics import normalize_claude_usage, normalize_codex_usage + + +def _breakdown( + *, + input_tokens: int, + cached_input_tokens: int, + output_tokens: int, + reasoning_output_tokens: int, + total_tokens: int, +): + return SimpleNamespace( + input_tokens=input_tokens, + cached_input_tokens=cached_input_tokens, + output_tokens=output_tokens, + reasoning_output_tokens=reasoning_output_tokens, + total_tokens=total_tokens, + ) + + +def test_claude_input_includes_cache_reads_and_writes(): + usage = normalize_claude_usage({ + "input_tokens": 100, + "cache_creation_input_tokens": 20, + "cache_read_input_tokens": 30, + "output_tokens": 40, + }, { + "claude-main": { + "contextWindow": 200_000, + "inputTokens": 150, + "outputTokens": 40, + }, + }) + + assert usage is not None + assert usage["input_tokens"] == 150 + assert usage["uncached_input_tokens"] == 100 + assert usage["cache_creation_input_tokens"] == 20 + assert usage["cache_read_input_tokens"] == 30 + assert usage["output_tokens"] == 40 + assert usage["total_tokens"] == 190 + assert usage["model_context_window"] == 200_000 + assert usage["provider_model_usage"]["claude-main"]["inputTokens"] == 150 + + +def test_codex_thread_delta_sums_every_model_call_in_the_turn(): + # The first notification says this thread had 800 tokens before the turn: + # first.total (1,100) - first.last (300). The final cumulative total is + # 1,900, so this turn processed 1,100 tokens across all model calls. + first = SimpleNamespace( + last=_breakdown( + input_tokens=200, + cached_input_tokens=100, + output_tokens=100, + reasoning_output_tokens=50, + total_tokens=300, + ), + total=_breakdown( + input_tokens=1_000, + cached_input_tokens=400, + output_tokens=100, + reasoning_output_tokens=50, + total_tokens=1_100, + ), + model_context_window=200_000, + ) + final = SimpleNamespace( + last=_breakdown( + input_tokens=300, + cached_input_tokens=150, + output_tokens=100, + reasoning_output_tokens=50, + total_tokens=400, + ), + total=_breakdown( + input_tokens=1_700, + cached_input_tokens=800, + output_tokens=200, + reasoning_output_tokens=100, + total_tokens=1_900, + ), + model_context_window=200_000, + ) + + usage = normalize_codex_usage(first, final) + + assert usage is not None + assert usage["calculation"] == "thread_delta" + assert usage["input_tokens"] == 900 + assert usage["cache_read_input_tokens"] == 500 + assert usage["uncached_input_tokens"] == 400 + assert usage["output_tokens"] == 200 + assert usage["reasoning_output_tokens"] == 100 + assert usage["total_tokens"] == 1_100 + assert usage["model_context_window"] == 200_000 + + +def test_codex_counter_reset_uses_labelled_latest_call_fallback(): + first = { + "last": { + "inputTokens": 100, + "cachedInputTokens": 50, + "outputTokens": 20, + "reasoningOutputTokens": 10, + "totalTokens": 120, + }, + "total": { + "inputTokens": 1_000, + "cachedInputTokens": 500, + "outputTokens": 200, + "reasoningOutputTokens": 100, + "totalTokens": 1_200, + }, + } + final = { + "last": { + "inputTokens": 80, + "cachedInputTokens": 40, + "outputTokens": 20, + "reasoningOutputTokens": 10, + "totalTokens": 100, + }, + # Lower than the inferred 1,080-token baseline: the thread counter reset. + "total": { + "inputTokens": 80, + "cachedInputTokens": 40, + "outputTokens": 20, + "reasoningOutputTokens": 10, + "totalTokens": 100, + }, + } + + usage = normalize_codex_usage(first, final) + + assert usage is not None + assert usage["calculation"] == "last_call_fallback" + assert usage["input_tokens"] == 80 + assert usage["total_tokens"] == 100 diff --git a/frontend/src/api/client.js b/frontend/src/api/client.js index 169ced25d..97fc27fe6 100644 --- a/frontend/src/api/client.js +++ b/frontend/src/api/client.js @@ -421,6 +421,10 @@ export const api = { method: 'POST', body: JSON.stringify({ activity_version: activityVersion }), }), + markPreviewSeen: (appId, updatedAt, final = false) => apiFetch(`/apps/${appId}/preview/seen`, { + method: 'POST', + body: JSON.stringify({ updated_at: updatedAt, final }), + }), update: (appId, payload) => listAffectingMutation('apps', `/apps/${appId}`, { method: 'PATCH', body: JSON.stringify(payload), diff --git a/frontend/src/components/ChatView/ActivityStretch.jsx b/frontend/src/components/ChatView/ActivityStretch.jsx index 42ff3fae7..340451055 100644 --- a/frontend/src/components/ChatView/ActivityStretch.jsx +++ b/frontend/src/components/ChatView/ActivityStretch.jsx @@ -1,6 +1,7 @@ -import { useId, useMemo, useRef } from 'react' +import { useEffect, useId, useMemo, useRef, useState } from 'react' import { StandardMarkdown } from './markdown/BlockRenderer.jsx' import ToolBlock from './ToolBlock.jsx' +import { apiFetch, jsonOrThrow } from '../../api/client.js' import { activityStreamState, activityDisplayState, @@ -175,7 +176,22 @@ function SingleActivity({ entry, chatId, live, surfaceKey }) { ) } -function GroupedActivityStretch({ entries, chatId, live = false, surfaceKey }) { +export function activityDetailUrl(chatId, detailRef) { + const messageIndex = encodeURIComponent(detailRef.message_index) + const start = encodeURIComponent(detailRef.start) + const end = encodeURIComponent(detailRef.end) + return `/chats/${encodeURIComponent(chatId)}/activity-detail` + + `?message_index=${messageIndex}&start=${start}&end=${end}` +} + +function GroupedActivityStretch({ + entries, + chatId, + live = false, + surfaceKey, + detailRef = null, + summaryToolCount = null, +}) { const stretchKey = assistantBlockKey(entries[0]?.item, entries[0]?.idx) const [userOpen, setUserOpen] = useDisclosureState( chatId, @@ -184,6 +200,56 @@ function GroupedActivityStretch({ entries, chatId, live = false, surfaceKey }) { const headerRef = useRef(null) const timelineRef = useRef(null) const timelineId = useId() + const [detailEntries, setDetailEntries] = useState(null) + const [detailError, setDetailError] = useState(false) + const [detailAttempt, setDetailAttempt] = useState(0) + const detailMessageIndex = detailRef?.message_index + const detailStart = detailRef?.start + const detailEnd = detailRef?.end + const detailKey = detailRef + ? `${detailMessageIndex}:${detailStart}:${detailEnd}` + : '' + + useEffect(() => { + setDetailEntries(null) + setDetailError(false) + }, [detailKey]) + + useEffect(() => { + if (!userOpen || !detailRef || detailEntries || detailError) return undefined + const controller = new AbortController() + let current = true + apiFetch(activityDetailUrl(chatId, { + message_index: detailMessageIndex, + start: detailStart, + end: detailEnd, + }), { + signal: controller.signal, + }) + .then(res => jsonOrThrow(res, 'Activity detail failed')) + .then(data => { + if (!current) return + setDetailEntries(Array.isArray(data.entries) ? data.entries : []) + }) + .catch(error => { + if (!current || error?.name === 'AbortError') return + setDetailError(true) + }) + return () => { + current = false + controller.abort() + } + }, [ + chatId, + detailAttempt, + detailEntries, + detailError, + detailEnd, + detailKey, + detailMessageIndex, + detailStart, + userOpen, + ]) const lastItem = entries[entries.length - 1]?.item const liveThinkingTail = live && lastItem?.type === 'thinking' @@ -245,8 +311,15 @@ function GroupedActivityStretch({ entries, chatId, live = false, surfaceKey }) { if (code != null && code !== 0) exitCode = code } } - return { state, exitCode, toolCount: tools.length, thinkingOnly: tools.length === 0 } - }, [sig]) // eslint-disable-line react-hooks/exhaustive-deps + return { + state, + exitCode, + toolCount: Number.isInteger(summaryToolCount) + ? summaryToolCount + : tools.length, + thinkingOnly: tools.length === 0, + } + }, [sig, summaryToolCount]) // eslint-disable-line react-hooks/exhaustive-deps const { state, exitCode, toolCount, thinkingOnly } = meta // The one presentation authority for icon, chip, and state class: a live @@ -278,6 +351,7 @@ function GroupedActivityStretch({ entries, chatId, live = false, surfaceKey }) { ? ', in progress' : '' const iconKind = thinkingOnly ? 'reasoning' : leadToolIcon + const timelineEntries = detailRef ? detailEntries : entries return (
))}