feat(craft): outputs reconciler wired into every terminal branch

Diffs the sandbox outputs manifest against the artifact rows at turn end,
while the turn still holds its prompt slot, so the next turn never starts
against a half-written index. Upserts only rows whose content or type moved,
keeping turn_index the turn that last changed each artifact, flags vanished
paths in one statement, and announces changed rows over Redis so an attached
stream can render artifact packets promptly. Incomplete manifests, truncated
or partially unreadable, are never reduced: the rows stay the baseline and
the next turn end self-heals. Wired into the interactive executor's owned
terminal branches and every owned scheduled-task terminal status, with the
docker manifest exec bounded by timeout(1) to match the kubernetes RPC
bound.

Author: nmgarza5

backend/onyx/server/features/build/db/artifact.py

@@ -9,7 +9,7 @@
 
 from uuid import UUID
 
-from sqlalchemy import case, desc, func, select
+from sqlalchemy import case, desc, func, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 
@@ -94,6 +94,35 @@ def mark_artifact_deleted(
     return artifact
 
 
+def mark_artifacts_deleted(
+    db_session: Session,
+    *,
+    session_id: UUID,
+    paths: list[str],
+) -> list[Artifact]:
+    """Flag every live row whose path vanished from the manifest, in one
+    statement. Returns the rows that actually flipped."""
+    if not paths:
+        return []
+    stmt = (
+        update(Artifact)
+        .where(
+            Artifact.session_id == session_id,
+            Artifact.path.in_(paths),
+            Artifact.deleted.is_(False),
+        )
+        .values(deleted=True, updated_at=func.now())
+        .returning(Artifact)
+    )
+    return list(
+        db_session.execute(
+            select(Artifact)
+            .from_statement(stmt)
+            .execution_options(populate_existing=True)
+        ).scalars()
+    )
+
+
 def get_session_artifacts(
     db_session: Session,
     *,

backend/onyx/server/features/build/interactive_turns/executor.py

@@ -33,6 +33,10 @@
     get_active_turn,
     touch_turn,
 )
+from onyx.server.features.build.outputs_reconciler import (
+    announce_artifacts,
+    reconcile_session_outputs,
+)
 from onyx.server.features.build.sandbox.event_schema import (
     ActivityTimeoutError,
     PromptResponse,
@@ -292,6 +296,24 @@ def persist_turn_error(message: str) -> None:
                     "Failed to persist turn error message for turn %s", turn_id
                 )
 
+        def reconcile_outputs() -> None:
+            """Best-effort outputs index update on every owned terminal branch,
+            while the prompt slot is still held. Failures never block
+            finish_turn."""
+            try:
+                packets = reconcile_session_outputs(
+                    db_session,
+                    get_sandbox_manager(),
+                    sandbox_id=sandbox.id,
+                    session_id=session_id,
+                    turn_index=turn_index,
+                )
+                db_session.commit()
+                announce_artifacts(session_id, packets, cache)
+            except Exception:
+                logger.exception("Outputs reconcile failed for turn %s", turn_id)
+                db_session.rollback()
+
         prompt_slot_cm = session_manager.prompt_slot(
             sandbox.id,
             session_id,
@@ -359,6 +381,7 @@ def persist_turn_error(message: str) -> None:
             if interrupt_requested():
                 session_manager.finalize_persist(session_id, state)
                 db_session.commit()
+                reconcile_outputs()
                 finish_turn(
                     cache=cache,
                     turn_id=turn_id,
@@ -414,6 +437,8 @@ def drive_one_prompt(
                         persist_turn_error(
                             "This turn was interrupted and could not finish."
                         )
+                        # No reconcile here: the lease is lost, so the slot
+                        # holder owns the index now.
                         finish_turn(
                             cache=cache,
                             turn_id=turn_id,
@@ -441,6 +466,7 @@ def drive_one_prompt(
                         session_manager.finalize_persist(session_id, state)
                         db_session.commit()
                         persist_turn_error(sandbox_event.message)
+                        reconcile_outputs()
                         finish_turn(
                             cache=cache,
                             turn_id=turn_id,
@@ -491,6 +517,7 @@ def drive_one_prompt(
 
             session_manager.finalize_persist(session_id, state)
             db_session.commit()
+            reconcile_outputs()
 
             if deadline_exceeded:
                 persist_turn_error(
@@ -543,6 +570,8 @@ def drive_one_prompt(
             except Exception:
                 logger.exception("Failed to finalize persistence for turn %s", turn_id)
             persist_turn_error("This turn failed unexpectedly.")
+            if not slot.lost:
+                reconcile_outputs()
             finish_turn(
                 cache=cache,
                 turn_id=turn_id,

backend/onyx/server/features/build/outputs_reconciler.py

@@ -0,0 +1,186 @@
+"""Reconciles a session's outputs tree into artifact rows at turn end.
+
+One call per owned terminal branch, while the turn still holds its prompt
+slot, so the next turn never starts against a half-written index. Idempotent
+because the rows are the baseline: a run skipped by a hard crash self-heals
+at the next turn end. An incomplete manifest is skipped outright, missing
+entries would flap rows deleted and directory hashes stale.
+"""
+
+from uuid import UUID
+
+from pydantic import ValidationError
+from sqlalchemy.orm import Session
+
+from onyx.cache.interface import CacheBackend
+from onyx.db.models import Artifact
+from onyx.server.features.build.artifact_classifier import (
+    OutputEntry,
+    derive_artifacts,
+)
+from onyx.server.features.build.db.artifact import (
+    get_session_artifacts,
+    mark_artifacts_deleted,
+    upsert_artifact,
+)
+from onyx.server.features.build.packets import ArtifactPacket
+from onyx.server.features.build.sandbox.base import SandboxManager
+from onyx.utils.logger import setup_logger
+
+logger = setup_logger()
+
+_ANNOUNCE_TTL_S = 60
+# The announce exists so an attached stream can render cards promptly. The
+# index refetch at turn end is the completeness guarantee, so a huge first
+# reconcile does not need every row on the wire.
+_MAX_ANNOUNCED = 50
+
+
+def _announce_key(session_id: UUID) -> str:
+    return f"craft:artifact:announce:{session_id}"
+
+
+def announce_artifacts(
+    session_id: UUID, packets: list[ArtifactPacket], cache: CacheBackend
+) -> None:
+    """Hand changed rows to the SSE stream attached to this session.
+
+    Called after commit, so a card never announces a row a reader cannot
+    fetch.
+    """
+    if not packets:
+        return
+    key = _announce_key(session_id)
+    for packet in packets[:_MAX_ANNOUNCED]:
+        cache.rpush(key, packet.model_dump_json())
+    cache.expire(key, _ANNOUNCE_TTL_S)
+
+
+def pop_artifact_announcement(
+    session_id: UUID, timeout_s: int, cache: CacheBackend
+) -> ArtifactPacket | None:
+    """BLPOP one announced artifact. None on timeout or unparseable payload."""
+    result = cache.blpop([_announce_key(session_id)], timeout_s)
+    if result is None:
+        return None
+    _key, value = result
+    if isinstance(value, bytes):
+        value = value.decode()
+    try:
+        return ArtifactPacket.model_validate_json(value)
+    except ValidationError:
+        logger.warning("artifact: unparseable announce %r for %s", value, session_id)
+        return None
+
+
+def reconcile_session_outputs(
+    db_session: Session,
+    sandbox_manager: SandboxManager,
+    *,
+    sandbox_id: UUID,
+    session_id: UUID,
+    turn_index: int | None,
+) -> list[ArtifactPacket]:
+    """Diff the sandbox outputs manifest against the artifact rows.
+
+    Upserts rows whose content or type moved or whose path came back, flags
+    rows whose path vanished, and leaves untouched rows alone so
+    ``turn_index`` keeps naming the turn that last changed each artifact.
+    Flushes only, the caller owns the commit and announces the returned
+    packets after it.
+
+    Returns an empty list without touching rows when the manifest is
+    unavailable or incomplete: the rows stay the baseline and the next turn
+    end self-heals. Unreadable entries count as incomplete because the
+    delete pass would otherwise flag rows the walk merely failed to see.
+    """
+    try:
+        manifest = sandbox_manager.get_outputs_manifest(
+            sandbox_id=sandbox_id, session_id=session_id
+        )
+    except (RuntimeError, ValidationError):
+        logger.warning(
+            "Outputs manifest unavailable for session %s; skipping reconcile",
+            session_id,
+            exc_info=True,
+        )
+        return []
+    if manifest.truncated or manifest.skipped_unreadable:
+        logger.warning(
+            "Outputs manifest incomplete for session %s "
+            "(truncated=%s skipped_unreadable=%d); skipping reconcile",
+            session_id,
+            manifest.truncated,
+            manifest.skipped_unreadable,
+        )
+        return []
+
+    derived = derive_artifacts(
+        [
+            OutputEntry(
+                path=entry.path,
+                is_directory=entry.is_directory,
+                size=entry.size,
+                mtime_ns=entry.mtime_ns,
+                sha256=entry.sha256,
+            )
+            for entry in manifest.entries
+        ]
+    )
+    rows_by_path = {
+        row.path: row
+        for row in get_session_artifacts(
+            db_session, session_id=session_id, include_deleted=True
+        )
+    }
+
+    packets: list[ArtifactPacket] = []
+    for artifact in derived:
+        row = rows_by_path.get(artifact.path)
+        unchanged = (
+            row is not None
+            and not row.deleted
+            and row.content_hash == artifact.content_hash
+            and row.type == artifact.type
+        )
+        if unchanged:
+            continue
+        updated = upsert_artifact(
+            db_session,
+            session_id=session_id,
+            artifact_type=artifact.type,
+            path=artifact.path,
+            name=artifact.name,
+            turn_index=turn_index,
+            size_bytes=artifact.size_bytes,
+            content_hash=artifact.content_hash,
+        )
+        packets.append(_packet_for(updated))
+
+    derived_paths = {artifact.path for artifact in derived}
+    vanished = [
+        path
+        for path, row in rows_by_path.items()
+        if path not in derived_paths and not row.deleted
+    ]
+    packets.extend(
+        _packet_for(row)
+        for row in mark_artifacts_deleted(
+            db_session, session_id=session_id, paths=vanished
+        )
+    )
+    return packets
+
+
+def _packet_for(row: Artifact) -> ArtifactPacket:
+    return ArtifactPacket(
+        artifact_id=row.id,
+        session_id=row.session_id,
+        path=row.path,
+        name=row.name,
+        artifact_type=row.type,
+        version=row.version,
+        turn_index=row.turn_index,
+        size_bytes=row.size_bytes,
+        deleted=row.deleted,
+    )

backend/onyx/server/features/build/packets.py

@@ -18,9 +18,8 @@
 - prompt_response: Agent finished processing
 - error: An error occurred
 
-Custom Onyx packets (defined here):
-- error: Onyx-specific errors (e.g., session not found)
-- subagent_started: A child opencode session was created under a parent turn
+The custom Onyx packet classes below are the authoritative list; the
+``BuildPacket`` union at the bottom names them all.
 """
 
 from datetime import datetime, timezone
@@ -29,6 +28,8 @@
 
 from pydantic import BaseModel, Field
 
+from onyx.db.enums import ArtifactType
+
 # =============================================================================
 # Base Packet Type
 # =============================================================================
@@ -89,6 +90,26 @@ class ConnectAppRequestPacket(BasePacket):
     reason: str | None = None
 
 
+class ArtifactPacket(BasePacket):
+    """An artifact row was produced, changed, or deleted at turn end.
+
+    Carries the full row, unlike the ids-only approval packet, so a consumer
+    can render a card with no round trip. The version is pinned at announce
+    time and the index refetch at turn end is the completeness guarantee.
+    """
+
+    type: Literal["artifact"] = "artifact"
+    artifact_id: UUID
+    session_id: UUID
+    path: str
+    name: str
+    artifact_type: ArtifactType
+    version: int
+    turn_index: int | None
+    size_bytes: int | None
+    deleted: bool
+
+
 class ContextUsagePacket(BasePacket):
     type: Literal["context_usage"] = "context_usage"
     used_tokens: int
@@ -109,6 +130,7 @@ class CompactionPacket(BasePacket):
     | ApprovalRequestedPacket
     | SubagentStartedPacket
     | ConnectAppRequestPacket
+    | ArtifactPacket
     | ContextUsagePacket
     | CompactionPacket
 )

backend/onyx/server/features/build/sandbox/docker/docker_sandbox_manager.py

@@ -1637,6 +1637,11 @@ def get_outputs_manifest(
             result = _run_in_container_as_sandbox_user(
                 container,
                 [
+                    # exec_run has no timeout, so a pathological walk must
+                    # not hang the turn's terminal handling. 30s matches the
+                    # kubernetes RPC bound.
+                    "/usr/bin/timeout",
+                    "30",
                     "/usr/local/bin/python3",
                     "-E",
                     "-s",