diff --git a/backend/app/artifacts.py b/backend/app/artifacts.py index e251d6d0c..a211d802e 100644 --- a/backend/app/artifacts.py +++ b/backend/app/artifacts.py @@ -24,10 +24,13 @@ from __future__ import annotations import hashlib import json import logging +import time from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path from typing import Any +from app.run_context import get_current_run_context from app.run_journal.models import CommittedRunEvent, RunEventDraft, RunRecord from app.run_journal.store import SQLiteRunJournal from app.utils.file_utils import list_files @@ -36,9 +39,18 @@ from app.utils.workspace_resolver import TaskSnapshot, get_workspace_resolver logger = logging.getLogger("artifacts") MAX_ARTIFACTS_PER_RUN = 500 +MAX_ARTIFACT_SCAN_SECONDS = 3.0 +MAX_ARTIFACT_SCAN_ENTRIES = 100_000 _ACTIVE_RUN_STATUSES = {"pending", "running", "waiting_for_user"} +@dataclass(frozen=True) +class ArtifactScanResult: + artifacts: list[dict[str, Any]] + scan_status: str + truncated: bool + + def _canonical_digest(value: Any) -> str: encoded = json.dumps( value, @@ -62,37 +74,64 @@ def _task_change_roots(snapshot: TaskSnapshot) -> list[tuple[Path, bool]]: return roots -def scan_task_changed_files( +def discover_task_changed_files( snapshot: TaskSnapshot, max_entries: int = MAX_ARTIFACTS_PER_RUN, modification_windows: tuple[tuple[float, float | None], ...] | None = None, *, list_files_fn: Callable[..., list[str]] = list_files, -) -> list[dict[str, Any]]: - """Discover files generated or modified by exactly one Run.""" +) -> ArtifactScanResult: + """Discover files generated or modified by one Run within hard budgets.""" result: list[dict[str, Any]] = [] seen_paths: set[str] = set() remaining = max_entries windows = modification_windows or ((snapshot.task_start_time - 1.0, None),) + deadline = time.perf_counter() + MAX_ARTIFACT_SCAN_SECONDS + scanned_entries = 0 + scan_limited = False + + def bounded_list_files( + root: Path, *, limit: int, **kwargs: Any + ) -> list[str]: + nonlocal scanned_entries, scan_limited + seconds_left = deadline - time.perf_counter() + entries_left = MAX_ARTIFACT_SCAN_ENTRIES - scanned_entries + if seconds_left <= 0 or entries_left <= 0: + scan_limited = True + return [] + stats: dict[str, float | int] = {} + values = list_files_fn( + str(root), + base=str(root), + # Read one look-ahead result so an exact result cap is not confused + # with a complete scan. + max_entries=limit + 1, + max_scanned_entries=entries_left, + max_scan_seconds=seconds_left, + stats=stats, + **kwargs, + ) + scanned_entries += int(stats.get("scanned_entries", 0)) + if bool(stats.get("scan_limited", 0)) or len(values) > limit: + scan_limited = True + return values[:limit] for root, include_all in _task_change_roots(snapshot): if remaining <= 0: + scan_limited = True break if include_all: - paths = list_files_fn( - str(root), base=str(root), max_entries=remaining - ) + paths = bounded_list_files(root, limit=remaining) else: paths = [] window_seen: set[str] = set() for modified_after, modified_before in windows: if len(paths) >= remaining: break - window_paths = list_files_fn( - str(root), - base=str(root), - max_entries=remaining - len(paths), + window_paths = bounded_list_files( + root, + limit=remaining - len(paths), modified_after=modified_after, modified_before=modified_before, ) @@ -138,7 +177,28 @@ def scan_task_changed_files( if remaining <= 0: break - return sorted(result, key=lambda item: item["relativePath"]) + return ArtifactScanResult( + artifacts=sorted(result, key=lambda item: item["relativePath"]), + scan_status="partial" if scan_limited else "complete", + truncated=scan_limited, + ) + + +def scan_task_changed_files( + snapshot: TaskSnapshot, + max_entries: int = MAX_ARTIFACTS_PER_RUN, + modification_windows: tuple[tuple[float, float | None], ...] | None = None, + *, + list_files_fn: Callable[..., list[str]] = list_files, +) -> list[dict[str, Any]]: + """Compatibility wrapper returning only the bounded Artifact list.""" + + return discover_task_changed_files( + snapshot, + max_entries=max_entries, + modification_windows=modification_windows, + list_files_fn=list_files_fn, + ).artifacts def task_modification_windows( @@ -190,6 +250,7 @@ def record_artifact_manifest( project_id: str, artifacts: list[dict[str, Any]], scan_status: str = "complete", + truncated: bool = False, ) -> CommittedRunEvent: """Commit Artifact lifecycle events followed by one manifest barrier.""" @@ -224,7 +285,7 @@ def record_artifact_manifest( "artifacts": projected, "artifact_count": len(projected), "scan_status": scan_status, - "truncated": len(projected) >= MAX_ARTIFACTS_PER_RUN, + "truncated": truncated, } manifest_digest = _canonical_digest( { @@ -264,31 +325,33 @@ def finalize_run_artifacts( ) -> CommittedRunEvent: """Discover and commit a Run manifest exactly before its terminal event.""" + current_run = journal.get_run(run.run_id) or run existing = journal.get_run_artifact_manifest_event(run.run_id) - if existing is not None: + if existing is not None and current_run.status in { + "completed", + "failed", + "cancelled", + }: return existing email: str | None = None user_id: str | int | None = None - try: - from app.service.task import get_task_lock_if_exists - - task_lock = get_task_lock_if_exists(run.project_id) - if task_lock is not None: - email = task_lock.email - user_id = task_lock.user_id - except Exception: - logger.exception( - "Failed to resolve Run workspace owner for Artifact finalization", - extra={"run_id": run.run_id}, - ) - + run_context = get_current_run_context() + if run_context is not None and run_context.run_id == run.run_id: + email = run_context.email + user_id = run_context.user_id + resolver = get_workspace_resolver() snapshot = ( - get_workspace_resolver().store.get_snapshot(email, run.run_id, user_id) + resolver.store.get_snapshot(email, run.run_id, user_id) if email else None ) - if snapshot is None or snapshot.project_id != run.project_id: + if snapshot is None: + located = resolver.store.find_snapshot(run.run_id) + if located is not None: + email, snapshot = located + user_id = snapshot.user_id + if snapshot is None: return record_artifact_manifest( journal, run_id=run.run_id, @@ -296,27 +359,44 @@ def finalize_run_artifacts( artifacts=[], scan_status="workspace_unavailable", ) + if snapshot.project_id != run.project_id: + return record_artifact_manifest( + journal, + run_id=run.run_id, + project_id=run.project_id, + artifacts=[], + scan_status="workspace_mismatch", + ) - if snapshot.artifact_manifest is not None: + if snapshot.artifact_manifest is not None and current_run.status in { + "completed", + "failed", + "cancelled", + }: artifacts = [dict(item) for item in snapshot.artifact_manifest] + scan_status = "complete" + truncated = False else: windows, _ = task_modification_windows( journal, run.run_id, run.project_id ) - artifacts = scan_task_changed_files( + scan_result = discover_task_changed_files( snapshot, modification_windows=windows ) + artifacts = scan_result.artifacts + scan_status = scan_result.scan_status + truncated = scan_result.truncated manifest = record_artifact_manifest( journal, run_id=run.run_id, project_id=run.project_id, artifacts=artifacts, + scan_status=scan_status, + truncated=truncated, ) try: - get_workspace_resolver().store.freeze_artifact_manifest( - email, snapshot, artifacts - ) + resolver.store.freeze_artifact_manifest(email, snapshot, artifacts) except Exception: # The sidecar snapshot is a compatibility cache. SQLite is already # authoritative and must not be rolled back by a cache write failure. diff --git a/backend/app/controller/run_controller.py b/backend/app/controller/run_controller.py index 68fa4a3bc..b8092fa58 100644 --- a/backend/app/controller/run_controller.py +++ b/backend/app/controller/run_controller.py @@ -22,13 +22,14 @@ depends on an in-memory queue. from __future__ import annotations import asyncio +import hashlib import json import logging import time import uuid from contextlib import suppress from dataclasses import asdict -from typing import Any +from typing import Annotated, Any from fastapi import APIRouter, Depends, Header, HTTPException, Query from fastapi.responses import StreamingResponse @@ -41,6 +42,7 @@ from app.run_journal import ( IdempotencyConflictError, InvalidRunTransitionError, OptimisticConcurrencyError, + RunEventDraft, RunNotFoundError, UnsafeResumeError, get_default_run_journal, @@ -107,6 +109,15 @@ class InteractionDecisionBody(BaseModel): continue_active_attempt: bool = True +class ArtifactUploadedBody(BaseModel): + chat_file_id: int | None = Field(default=None, ge=1) + s3_bucket: str = Field(min_length=1, max_length=255) + s3_key: str = Field(min_length=1, max_length=2048) + filename: str = Field(min_length=1, max_length=1024) + file_size: int = Field(ge=0) + file_type: str = Field(default="", max_length=255) + + def _event_payload(event: CommittedRunEvent) -> dict[str, Any]: return { "event_id": event.event_id, @@ -300,7 +311,7 @@ def _total_attempt_elapsed_ms(attempts: list[Any], *, now: float) -> int: @router.get("/runs") async def list_project_runs( project_id: str = Query(min_length=1), - status: list[str] | None = Query(default=None), + status: Annotated[list[str] | None, Query()] = None, limit: int = Query(default=20, ge=1, le=100), ): """Return canonical Run state for the main Desktop Project UI.""" @@ -677,6 +688,84 @@ async def cancel_run(run_id: str, body: CancelRunBody): return asdict(run) +@router.post("/runs/{run_id}/artifacts/{artifact_id}/uploaded") +async def record_artifact_uploaded( + run_id: str, + artifact_id: str, + body: ArtifactUploadedBody, +): + """Attach a durable Cloud asset reference to one canonical Artifact.""" + + journal = get_default_run_journal() + manifest = await asyncio.to_thread( + journal.get_run_artifact_manifest_event, + run_id, + ) + if manifest is None: + raise HTTPException( + status_code=409, detail="Artifact manifest missing" + ) + candidates = manifest.payload.get("artifacts") + artifact = ( + next( + ( + item + for item in candidates + if isinstance(item, dict) + and item.get("artifact_id") == artifact_id + ), + None, + ) + if isinstance(candidates, list) + else None + ) + if artifact is None: + raise HTTPException(status_code=404, detail="Artifact not found") + if artifact.get("uploadPolicy") != "agent_generated": + raise HTTPException( + status_code=409, + detail="Metadata-only Artifact cannot be uploaded automatically", + ) + + payload = { + "artifact_id": artifact_id, + "relativePath": artifact.get("relativePath"), + "filename": body.filename, + "asset_ref": { + "chat_file_id": body.chat_file_id, + "bucket": body.s3_bucket, + "key": body.s3_key, + "filename": body.filename, + "size": body.file_size, + "content_type": body.file_type, + }, + } + event_fingerprint = hashlib.sha256( + json.dumps( + {"run_id": run_id, **payload}, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + try: + committed = await asyncio.to_thread( + journal.append_event, + run_id, + RunEventDraft( + event_id=f"au_{event_fingerprint[:61]}", + event_type="artifact.uploaded", + payload=payload, + ), + ) + except Exception as exc: + raise _control_error(exc) from exc + from app.run_sync.runtime import notify_default_cloud_sync_worker + + notify_default_cloud_sync_worker() + return _event_payload(committed) + + @router.post("/runs/{run_id}/signals") async def signal_run(run_id: str, body: RunSignalBody): journal = get_default_run_journal() diff --git a/backend/app/run_journal/store.py b/backend/app/run_journal/store.py index 781406176..1f37f084a 100644 --- a/backend/app/run_journal/store.py +++ b/backend/app/run_journal/store.py @@ -7692,10 +7692,10 @@ class SQLiteRunJournal: """Commit one Run's Artifact lifecycle and manifest barrier once. Artifact discovery runs before this transaction and two terminal paths - may race to publish its result. ``BEGIN IMMEDIATE`` plus the in-lock - barrier lookup makes the first finalized manifest authoritative; a - later scanner observes and returns it without appending a second view - of the same Run's filesystem. + may race to publish its result. While a Run remains non-terminal a + later scan may append a corrected manifest (for example after crash + recovery). Once terminal, ``BEGIN IMMEDIATE`` plus the in-lock barrier + lookup freezes and returns the latest committed manifest. """ if ( @@ -7723,7 +7723,16 @@ class SQLiteRunJournal: """, (run_id,), ).fetchone() - if existing is not None: + run = connection.execute( + "SELECT status FROM runs WHERE run_id = ?", (run_id,) + ).fetchone() + if run is None: + raise RunNotFoundError(f"run_id {run_id!r} does not exist") + if existing is not None and run["status"] in { + "completed", + "failed", + "cancelled", + }: return self._event_from_row(existing) committed: list[CommittedRunEvent] = [] @@ -7738,6 +7747,84 @@ class SQLiteRunJournal: ) return committed[-1] + def complete_successful_run( + self, + run_id: str, + *, + assistant_final: RunEventDraft, + terminal: RunEventDraft, + artifact_manifest: CommittedRunEvent, + expected_project_id: str, + ) -> tuple[CommittedRunEvent, CommittedRunEvent]: + """Atomically commit the canonical result and successful Run terminal. + + Artifact discovery intentionally happens before this short transaction. + A crash after its manifest barrier is safe because a non-terminal Run may + publish a later manifest generation; the successful assistant result and + ``run.completed`` themselves never become separated. + """ + + if assistant_final.event_type != "assistant.final": + raise ValueError("successful completion requires assistant.final") + if terminal.event_type != "run.completed": + raise ValueError("successful completion requires run.completed") + if artifact_manifest.run_id != run_id: + raise IdempotencyConflictError( + "Artifact manifest belongs to another Run" + ) + + terminal_payload = { + **dict(terminal.payload), + "artifact_manifest_event_id": artifact_manifest.event_id, + "artifact_count": int( + artifact_manifest.payload.get("artifact_count", 0) + ), + "result_event_id": assistant_final.event_id, + } + terminal_draft = RunEventDraft( + event_id=terminal.event_id, + event_type=terminal.event_type, + payload=terminal_payload, + legacy_step=terminal.legacy_step, + created_at=terminal.created_at, + ) + with self._write_transaction() as connection: + run = connection.execute( + "SELECT * FROM runs WHERE run_id = ?", (run_id,) + ).fetchone() + if run is None: + raise RunNotFoundError(f"run_id {run_id!r} does not exist") + if run["project_id"] != expected_project_id: + raise IdempotencyConflictError( + f"run_id {run_id!r} belongs to another project" + ) + + result_event = self._append_event_in_transaction( + connection, + run_id, + assistant_final, + expected_project_id=expected_project_id, + ) + terminal_event = self._append_event_in_transaction( + connection, + run_id, + terminal_draft, + expected_project_id=expected_project_id, + run_status="completed", + clear_active_attempt=True, + ) + connection.execute( + """ + UPDATE run_attempts + SET status = 'completed', ended_at = COALESCE(ended_at, ?), + outcome = COALESCE(outcome, 'run.completed') + WHERE run_id = ? + AND status IN ('pending', 'running', 'waiting_for_user') + """, + (terminal_draft.created_at, run_id), + ) + return result_event, terminal_event + def list_events( self, run_id: str, diff --git a/backend/app/run_runtime/coordinator.py b/backend/app/run_runtime/coordinator.py index 43173c569..6d194727a 100644 --- a/backend/app/run_runtime/coordinator.py +++ b/backend/app/run_runtime/coordinator.py @@ -368,14 +368,21 @@ class RunCoordinator: handle.deadline_changed_event.set() return True - async def complete_turn(self, run_id: str) -> bool: + async def complete_turn( + self, + run_id: str, + *, + project_id: str, + assistant_data: Any, + ) -> bool: """Terminalize one Run without disposing its warm Project runtime. Compatibility chat generators intentionally stay alive across follow-up Runs. Their physical completion therefore cannot define a - logical Run boundary. The end-step recorder calls this method after - assistant.final is durable; the same handle can then be rebound to - the next Run without retaining the previous Run as ``running``. + logical Run boundary. The end-step recorder gives this method the + assistant result so it can atomically commit that result with the + successful terminal; the same handle can then be rebound to the next + Run without retaining the previous Run as ``running``. """ async with self._lock: @@ -383,12 +390,70 @@ class RunCoordinator: if handle is None or not handle.consumer_alive: return False started_at = handle.started_at - await self._commit_run_terminal( - run_id=run_id, - started_at=started_at, - event_type="run.completed", - payload={"reason": "run_turn_completed"}, + if self._journal is None: + return True + from app.artifacts import finalize_run_artifacts + from app.run_journal.models import RunEventDraft + + run = await asyncio.to_thread(self._journal.get_run, run_id) + if run is None: + return False + artifact_manifest = await asyncio.to_thread( + finalize_run_artifacts, + self._journal, + run, ) + assistant_payload = ( + dict(assistant_data) + if isinstance(assistant_data, dict) + else {"message": str(assistant_data)} + ) + await asyncio.to_thread( + self._journal.complete_successful_run, + run_id, + assistant_final=RunEventDraft( + event_id=f"assistant-final:{run_id}", + event_type="assistant.final", + payload=assistant_payload, + legacy_step="end", + ), + terminal=RunEventDraft( + event_id=( + f"runtime-terminal:{run_id}:" + f"{int(started_at * 1_000_000)}:run.completed" + ), + event_type="run.completed", + payload={"reason": "run_turn_completed"}, + ), + artifact_manifest=artifact_manifest, + expected_project_id=project_id, + ) + from app.run_sync.runtime import notify_default_cloud_sync_worker + + notify_default_cloud_sync_worker() + try: + from app.workspace_git import get_default_workspace_git_lifecycle + + await asyncio.to_thread( + get_default_workspace_git_lifecycle().finalize_run, + run_id, + ) + except Exception: + logger.exception( + "Terminal Run Git finalization needs attention", + extra={"run_id": run_id}, + ) + try: + from app.lightweight_memory import ( + schedule_project_memory_maintenance, + ) + + schedule_project_memory_maintenance(project_id) + except Exception: + logger.exception( + "Failed to schedule non-blocking Memory maintenance", + extra={"run_id": run_id, "project_id": project_id}, + ) run = ( await asyncio.to_thread(self._journal.get_run, run_id) if self._journal is not None @@ -627,7 +692,12 @@ class RunCoordinator: if run is None or run.status in {"completed", "failed", "cancelled"}: return try: - if event_type == "run.completed": + if event_type in { + "run.completed", + "run.failed", + "run.cancelled", + "run.deadline_reached", + }: from app.artifacts import finalize_run_artifacts artifact_manifest_event = await asyncio.to_thread( @@ -635,10 +705,6 @@ class RunCoordinator: self._journal, run, ) - result_event = await asyncio.to_thread( - self._journal.get_run_final_result_event, - run_id, - ) payload = { **payload, "artifact_manifest_event_id": ( @@ -650,6 +716,10 @@ class RunCoordinator: ) ), } + result_event = await asyncio.to_thread( + self._journal.get_run_final_result_event, + run_id, + ) if result_event is not None: payload = { **payload, diff --git a/backend/app/run_sync/cloud_sync.py b/backend/app/run_sync/cloud_sync.py index 05e1af804..b335342f1 100644 --- a/backend/app/run_sync/cloud_sync.py +++ b/backend/app/run_sync/cloud_sync.py @@ -164,6 +164,12 @@ class RunEventSyncTransport(Protocol): payload: dict[str, Any], ) -> dict[str, Any]: ... + async def claim_memory_writer( + self, + configuration: CloudSyncConfiguration, + payload: dict[str, Any], + ) -> dict[str, Any]: ... + async def close(self) -> None: ... @@ -315,7 +321,7 @@ class HttpRunEventSyncTransport: return await self._json_request( "GET", f"{self._sync_base(configuration)}/projects/{encoded_project_id}/snapshot" - "?event_limit=1", + "?event_limit=1&include_artifacts=false", configuration, ) @@ -372,6 +378,24 @@ class HttpRunEventSyncTransport: payload, ) + async def claim_memory_writer( + self, + configuration: CloudSyncConfiguration, + payload: dict[str, Any], + ) -> dict[str, Any]: + if str(payload.get("scope_type")) == "project": + await self._ensure_device_and_route( + configuration, str(payload["scope_id"]) + ) + else: + await self._ensure_device(configuration) + return await self._json_request( + "POST", + f"{self._sync_base(configuration)}/memory/writer:claim", + configuration, + payload, + ) + async def close(self) -> None: await self._client.aclose() @@ -661,6 +685,53 @@ class CloudSyncWorker: } try: response = await put_snapshot(configuration, payload) + except RunEventSyncHttpError as exc: + detail = ( + exc.detail.get("detail", exc.detail) + if isinstance(exc.detail, dict) + else {} + ) + claim_writer = getattr( + self._transport, "claim_memory_writer", None + ) + if ( + exc.status_code == 409 + and isinstance(detail, dict) + and detail.get("code") == "memory_scope_writer_conflict" + and isinstance(detail.get("current_writer_epoch"), int) + and callable(claim_writer) + ): + try: + await claim_writer( + configuration, + { + "scope_type": key[0], + "scope_id": key[1], + "expected_writer_epoch": detail[ + "current_writer_epoch" + ], + }, + ) + response = await put_snapshot(configuration, payload) + except Exception: + logger.exception( + "Cloud Memory writer transfer failed for %s/%s", + *key, + ) + continue + else: + logger.exception( + "Cloud Memory snapshot sync failed for %s/%s", *key + ) + continue + except asyncio.CancelledError: + raise + except Exception: + logger.exception( + "Cloud Memory snapshot sync failed for %s/%s", *key + ) + continue + try: if ( response.get("scope_type") != key[0] or response.get("scope_id") != key[1] diff --git a/backend/app/utils/file_utils.py b/backend/app/utils/file_utils.py index d547c25c7..202311b1b 100644 --- a/backend/app/utils/file_utils.py +++ b/backend/app/utils/file_utils.py @@ -199,6 +199,8 @@ def list_files( skip_prefix: str = ".", modified_after: float | None = None, modified_before: float | None = None, + max_scanned_entries: int | None = None, + max_scan_seconds: float | None = None, stats: dict[str, float | int] | None = None, ) -> list[str]: """List files under dir_path with optional base confinement and filters. @@ -218,6 +220,9 @@ def list_files( is at or before this Unix timestamp. This lets historical Run artifact queries exclude files written by later Runs that share a direct-write workspace. + max_scanned_entries (int | None): Optional hard cap on filesystem + entries visited, including entries filtered out by mtime. + max_scan_seconds (float | None): Optional wall-clock scan budget. Returns: List of real absolute file paths under dir_path (subject to filters and max_entries). @@ -245,6 +250,8 @@ def list_files( scan_started = time.perf_counter() realpath_elapsed = 0.0 symlink_count = 0 + scanned_entries = 0 + scan_limited = False def record_stats() -> None: if stats is None: @@ -252,20 +259,49 @@ def list_files( stats["scan_elapsed_ms"] = (time.perf_counter() - scan_started) * 1000 stats["realpath_elapsed_ms"] = realpath_elapsed * 1000 stats["symlink_count"] = symlink_count + stats["scanned_entries"] = scanned_entries + stats["scan_limited"] = int(scan_limited) + + def budget_exhausted() -> bool: + nonlocal scan_limited + if ( + max_scanned_entries is not None + and scanned_entries >= max_scanned_entries + ): + scan_limited = True + return True + if ( + max_scan_seconds is not None + and time.perf_counter() - scan_started >= max_scan_seconds + ): + scan_limited = True + return True + return False try: for root, dirs, files in os.walk(resolved_dir, followlinks=False): + scanned_entries += 1 + len(dirs) + if budget_exhausted(): + record_stats() + return result dirs[:] = [ d for d in dirs if d not in skip_dirs and not _should_skip(d, skip_prefix) ] for name in files: + scanned_entries += 1 + if budget_exhausted(): + record_stats() + return result if _should_skip(name, skip_prefix, skip_extensions): continue try: file_path = os.path.join(root, name) - if modified_after is not None or modified_before is not None: + if ( + modified_after is not None + or modified_before is not None + ): file_mtime = os.stat( file_path, follow_symlinks=False ).st_mtime diff --git a/backend/app/utils/server/sync_step.py b/backend/app/utils/server/sync_step.py index 7c324b0ae..49941d532 100644 --- a/backend/app/utils/server/sync_step.py +++ b/backend/app/utils/server/sync_step.py @@ -198,27 +198,16 @@ async def _record_local_step(args, value) -> None: # was already shown to the user. await _flush_local_text(run_id) if data["step"] == "end": - # Finalize the authoritative Artifact manifest before the legacy - # assistant.final/end projection. Durable replay can therefore build - # the Files changed UI before END terminalizes the legacy reducer. - from app.artifacts import finalize_run_artifacts - from app.run_journal.runtime import get_default_run_journal - - journal = get_default_run_journal() - run = await asyncio.to_thread(journal.get_run, run_id) - if run is None: - raise RuntimeError( - f"Run {run_id!r} disappeared before finalization" - ) - await asyncio.to_thread(finalize_run_artifacts, journal, run) - await get_default_event_recorder().record_assistant_final( - project_id=project_id, - run_id=run_id, - data=data["data"], - ) + # RunCoordinator owns the successful terminal transaction. Artifact + # discovery happens first, then assistant.final + run.completed commit + # atomically before the legacy END frame is yielded to the Renderer. from app.run_runtime import get_default_run_coordinator - if not await get_default_run_coordinator().complete_turn(run_id): + if not await get_default_run_coordinator().complete_turn( + run_id, + project_id=project_id, + assistant_data=data["data"], + ): raise RuntimeError( f"RunCoordinator could not terminalize completed Run {run_id!r}" ) @@ -235,6 +224,11 @@ async def _record_local_step_fail_open(args, value) -> None: try: await _record_local_step(args, value) except Exception as exc: + parsed = _parse_value(value) + if parsed is not None and parsed.get("step") == "end": + # A successful END is a product claim that must never outrun the + # canonical assistant result and Run terminal transaction. + raise run_id, project_id = _resolve_run_and_project(args) _local_text_buffers.pop(run_id, None) _mark_local_history_degraded( diff --git a/backend/app/utils/workspace_resolver.py b/backend/app/utils/workspace_resolver.py index 0ab7fac1a..160831054 100644 --- a/backend/app/utils/workspace_resolver.py +++ b/backend/app/utils/workspace_resolver.py @@ -34,6 +34,7 @@ from app.utils.workspace_paths import ( project_task_root, project_workdir_root, run_output_root, + runtime_owner_key, workspace_state_root, ) @@ -366,26 +367,103 @@ class WorkspaceStore: ) return list(bindings_by_space.values()) + @staticmethod + def _read_snapshot(path: Path) -> TaskSnapshot | None: + try: + data = json.loads(path.read_text(encoding="utf-8")) + if "space_id" not in data: + data["space_id"] = data.get("project_id", "") + data.setdefault("user_id", None) + data.setdefault("workdir_mode", None) + data.setdefault("base_snapshot_id", None) + manifest = data.setdefault("artifact_manifest", None) + if manifest is not None: + data["artifact_manifest"] = tuple(manifest) + data.setdefault("artifacts_frozen_at", None) + return TaskSnapshot(**data) + except Exception: + logger.warning("Failed to read task snapshot: %s", path) + return None + def get_snapshot( self, email: str, task_id: str, user_id: str | int | None = None ) -> TaskSnapshot | None: for path in self._task_paths(email, task_id, user_id): if not path.exists(): continue - try: - data = json.loads(path.read_text(encoding="utf-8")) - if "space_id" not in data: - data["space_id"] = data.get("project_id", "") - data.setdefault("user_id", None) - data.setdefault("workdir_mode", None) - data.setdefault("base_snapshot_id", None) - manifest = data.setdefault("artifact_manifest", None) - if manifest is not None: - data["artifact_manifest"] = tuple(manifest) - data.setdefault("artifacts_frozen_at", None) - return TaskSnapshot(**data) - except Exception: - logger.warning("Failed to read task snapshot: %s", path) + snapshot = self._read_snapshot(path) + if snapshot is not None: + return snapshot + return None + + def find_snapshot(self, task_id: str) -> tuple[str, TaskSnapshot] | None: + """Resolve an immutable Run snapshot without mutable TaskLock state. + + Startup reconciliation has no active RunContext and must still be able + to finalize the Run that owned a workspace. Task ids are generated + identifiers, so reject path-shaped input before performing the bounded + one-level owner lookup under ``~/.eigent/workspaces``. + """ + + if ( + not task_id + or Path(task_id).name != task_id + or task_id in {".", ".."} + ): + return None + workspaces_root = workspace_state_root("_snapshot_lookup_").parent + if not workspaces_root.exists(): + return None + + matches: list[tuple[str, TaskSnapshot]] = [] + try: + owner_roots = list(workspaces_root.iterdir()) + except OSError: + logger.warning("Failed to enumerate workspace snapshot owners") + return None + for owner_root in owner_roots: + if not owner_root.is_dir(): + continue + path = owner_root / "tasks" / f"{task_id}.json" + if not path.is_file(): + continue + snapshot = self._read_snapshot(path) + if snapshot is not None and snapshot.task_id == task_id: + matches.append((owner_root.name, snapshot)) + + if not matches: + return None + if len(matches) == 1: + return matches[0] + + # A user-id snapshot can coexist briefly with its legacy email-keyed + # copy during migration. Prefer the canonical user owner only when all + # copies describe the same immutable workspace; conflicting copies are + # never guessed between. + identity = { + ( + item.project_id, + item.space_id, + item.user_id, + item.working_directory, + item.task_output_root, + ) + for _, item in matches + } + if len(identity) == 1: + snapshot = matches[0][1] + if snapshot.user_id is not None: + canonical_owner = runtime_owner_key("", snapshot.user_id) + for owner, item in matches: + if owner == canonical_owner: + return owner, item + return matches[0] + + logger.error( + "Ambiguous workspace snapshots for Run %s across owners %s", + task_id, + [owner for owner, _ in matches], + ) return None def save_snapshot(self, email: str, snapshot: TaskSnapshot) -> None: diff --git a/backend/tests/app/controller/test_run_controller.py b/backend/tests/app/controller/test_run_controller.py index e36f1d642..c70c25325 100644 --- a/backend/tests/app/controller/test_run_controller.py +++ b/backend/tests/app/controller/test_run_controller.py @@ -21,9 +21,11 @@ from unittest.mock import MagicMock, patch import pytest from app.controller.run_controller import ( + ArtifactUploadedBody, _is_terminal, get_run, get_run_events, + record_artifact_uploaded, stream_run_events, ) from app.run_journal import CommittedRunEvent, RunRecord @@ -87,6 +89,65 @@ def test_assistant_final_renders_as_legacy_end_without_closing_run_stream(): assert _is_terminal(event) is False +@pytest.mark.asyncio +async def test_uploaded_asset_is_journaled_only_for_agent_generated_artifact(): + journal = MagicMock() + journal.get_run_artifact_manifest_event.return_value = CommittedRunEvent( + event_id="manifest", + run_id="run-1", + sequence=2, + event_type="artifact.manifest.finalized", + payload={ + "artifacts": [ + { + "artifact_id": "art-1", + "relativePath": "report.csv", + "uploadPolicy": "agent_generated", + } + ] + }, + legacy_step=None, + created_at=1.0, + run_version=2, + ) + journal.append_event.side_effect = lambda run_id, draft: CommittedRunEvent( + event_id=draft.event_id, + run_id=run_id, + sequence=3, + event_type=draft.event_type, + payload=draft.payload, + legacy_step=None, + created_at=draft.created_at, + run_version=3, + ) + with ( + patch( + "app.controller.run_controller.get_default_run_journal", + return_value=journal, + ), + patch( + "app.run_sync.runtime.notify_default_cloud_sync_worker" + ) as notify, + ): + result = await record_artifact_uploaded( + "run-1", + "art-1", + ArtifactUploadedBody( + s3_bucket="generated", + s3_key="user/project/report.csv", + filename="report.csv", + file_size=12, + file_type="text/csv", + ), + ) + + draft = journal.append_event.call_args.args[1] + assert draft.event_type == "artifact.uploaded" + assert draft.payload["asset_ref"]["key"] == "user/project/report.csv" + assert result["event_type"] == "artifact.uploaded" + notify.assert_called_once_with() + + def _decode_sse(value: str) -> tuple[int | None, str, dict]: event_id = None event_name = "" @@ -228,9 +289,11 @@ async def test_stream_resumes_from_last_event_id_on_transport_reconnect(): events = [_event(1, "confirmed"), _event(2, "end")] journal = MagicMock() journal.get_run.return_value = _run_record() - journal.list_events.side_effect = lambda run_id, *, after_sequence, limit: [ - event for event in events if event.sequence > after_sequence - ][:limit] + journal.list_events.side_effect = ( + lambda run_id, *, after_sequence, limit: [ + event for event in events if event.sequence > after_sequence + ][:limit] + ) coordinator = RunCoordinator() with ( diff --git a/backend/tests/app/run_runtime/test_coordinator.py b/backend/tests/app/run_runtime/test_coordinator.py index fdc1a6c92..ddcbab800 100644 --- a/backend/tests/app/run_runtime/test_coordinator.py +++ b/backend/tests/app/run_runtime/test_coordinator.py @@ -266,14 +266,17 @@ async def test_logical_turn_completes_without_disposing_warm_runtime(tmp_path): run_id="run-turn", stream_factory=source, ) - final = await EventRecorder(journal).record_assistant_final( - project_id="project-1", - run_id="run-turn", - data="Done", + assert ( + await coordinator.complete_turn( + "run-turn", + project_id="project-1", + assistant_data="Done", + ) + is True ) - - assert await coordinator.complete_turn("run-turn") is True assert journal.get_run("run-turn").status == "completed" + final = journal.get_run_final_result_event("run-turn") + assert final is not None completed = next( event for event in journal.list_events("run-turn") diff --git a/backend/tests/app/run_sync/test_cloud_sync.py b/backend/tests/app/run_sync/test_cloud_sync.py index dbd06d612..8bcf6d6d7 100644 --- a/backend/tests/app/run_sync/test_cloud_sync.py +++ b/backend/tests/app/run_sync/test_cloud_sync.py @@ -44,6 +44,8 @@ class FakeTransport: self.memory_snapshots: list[dict[str, Any]] = [] self.memory_payloads: list[dict[str, Any]] = [] self.memory_snapshot_failures: set[tuple[str, str]] = set() + self.memory_writer_conflicts: set[tuple[str, str]] = set() + self.memory_writer_claims: list[dict[str, Any]] = [] async def ingest(self, configuration, payload): self.payloads.append(payload) @@ -108,6 +110,18 @@ class FakeTransport: async def put_memory_snapshot(self, configuration, payload): self.memory_snapshots.append(payload) + key = (payload["scope_type"], payload["scope_id"]) + if key in self.memory_writer_conflicts: + self.memory_writer_conflicts.remove(key) + raise RunEventSyncHttpError( + 409, + { + "detail": { + "code": "memory_scope_writer_conflict", + "current_writer_epoch": 4, + } + }, + ) if (payload["scope_type"], payload["scope_id"]) in ( self.memory_snapshot_failures ): @@ -119,6 +133,15 @@ class FakeTransport: "entry_count": len(payload["entries"]), } + async def claim_memory_writer(self, configuration, payload): + self.memory_writer_claims.append(payload) + return { + **payload, + "writer_epoch": payload["expected_writer_epoch"] + 1, + "owner_device_id": configuration.desktop_instance_id, + "rebase_required": True, + } + async def ingest_memory_mutations(self, configuration, payload): self.memory_payloads.append(payload) return { @@ -264,6 +287,43 @@ async def test_bad_memory_snapshot_does_not_block_an_unrelated_scope(journal): await worker.close() +@pytest.mark.asyncio +async def test_worker_claims_stale_memory_writer_then_retries_full_snapshot( + journal, +): + journal.apply_memory_mutation( + mutation_id="mutation-1", + idempotency_key="request-1", + operation="add", + scope_type="project", + scope_id="project-1", + memory_id="memory-1", + actor_type="user", + reason="Created in Memory Center", + content="Use Chinese.", + kind="preference", + token_count=3, + created_by="user", + source_trust="user_confirmed", + ) + transport = FakeTransport() + transport.memory_writer_conflicts.add(("project", "project-1")) + worker = _worker(journal, transport) + + assert await worker.drain_once() == 1 + + assert transport.memory_writer_claims == [ + { + "scope_type": "project", + "scope_id": "project-1", + "expected_writer_epoch": 4, + } + ] + assert len(transport.memory_snapshots) == 2 + assert len(transport.memory_payloads) == 1 + await worker.close() + + @pytest.mark.asyncio async def test_worker_redacts_approval_arguments_and_local_targets(journal): journal.ensure_run(run_id="run-approval", project_id="project-1", now=1) @@ -609,5 +669,8 @@ async def test_http_transport_uses_device_auth_for_history_bootstrap(): "/projects/project%2Fone/snapshot" in str(request.url) for request in requests ) - assert any("event_limit=1" in str(request.url) for request in requests) + assert any( + "event_limit=1&include_artifacts=false" in str(request.url) + for request in requests + ) await transport.close() diff --git a/backend/tests/app/test_artifacts.py b/backend/tests/app/test_artifacts.py index e84a52a9e..e0e8d7816 100644 --- a/backend/tests/app/test_artifacts.py +++ b/backend/tests/app/test_artifacts.py @@ -21,7 +21,7 @@ from app import artifacts from app.run_journal import SQLiteRunJournal -def test_finalize_commits_artifacts_then_manifest_and_is_idempotent( +def test_finalize_rescans_non_terminal_run_and_reuses_terminal_manifest( monkeypatch, tmp_path ): output_root = tmp_path / "output" @@ -43,39 +43,50 @@ def test_finalize_commits_artifacts_then_manifest_and_is_idempotent( working_directory=str(workspace_root), task_start_time=0, artifact_manifest=None, + user_id="user-1", ) resolver = MagicMock() - resolver.store.get_snapshot.return_value = snapshot + resolver.store.find_snapshot.return_value = ( + "user_user-1", + snapshot, + ) monkeypatch.setattr( artifacts, "get_workspace_resolver", lambda: resolver ) - from app.service import task as task_service - - monkeypatch.setattr( - task_service, - "get_task_lock_if_exists", - lambda _project_id: SimpleNamespace( - email="user@example.com", user_id="user-1" + first = artifacts.finalize_run_artifacts(journal, run) + resumed = output_root / "resumed.txt" + resumed.write_text( + "created after the first manifest", encoding="utf-8" + ) + second = artifacts.finalize_run_artifacts(journal, run) + journal.append_event( + "run-1", + artifacts.RunEventDraft( + event_id="run-1-completed", + event_type="run.completed", + payload={"artifact_manifest_event_id": second.event_id}, ), ) - - first = artifacts.finalize_run_artifacts(journal, run) - second = artifacts.finalize_run_artifacts(journal, run) + third = artifacts.finalize_run_artifacts(journal, run) events = journal.list_events("run-1") - assert first.event_id == second.event_id - assert {event.event_type for event in events[:-1]} == { + assert first.event_id != second.event_id + assert second.event_id == third.event_id + assert first.payload["artifact_count"] == 2 + assert second.payload["artifact_count"] == 3 + assert {event.event_type for event in events} >= { "artifact.created", "artifact.modified", + "artifact.manifest.finalized", + "run.completed", } - assert events[-1].event_type == "artifact.manifest.finalized" - assert first.payload["artifact_count"] == 2 - assert first.payload["scan_status"] == "complete" + assert second.payload["scan_status"] == "complete" assert { - artifact["uploadPolicy"] for artifact in first.payload["artifacts"] + artifact["uploadPolicy"] + for artifact in second.payload["artifacts"] } == {"agent_generated", "metadata_only"} - resolver.store.freeze_artifact_manifest.assert_called_once() + assert resolver.store.freeze_artifact_manifest.call_count == 2 finally: journal.close() @@ -86,10 +97,10 @@ def test_finalize_records_explicit_unavailable_manifest_without_workspace( journal = SQLiteRunJournal(tmp_path / "journal.sqlite3") try: run = journal.ensure_run(run_id="run-1", project_id="project-1") - from app.service import task as task_service - + resolver = MagicMock() + resolver.store.find_snapshot.return_value = None monkeypatch.setattr( - task_service, "get_task_lock_if_exists", lambda _project_id: None + artifacts, "get_workspace_resolver", lambda: resolver ) manifest = artifacts.finalize_run_artifacts(journal, run) @@ -113,6 +124,14 @@ def test_concurrent_manifest_finalization_commits_one_authoritative_barrier( barrier = Barrier(2) try: journal.ensure_run(run_id="run-1", project_id="project-1") + journal.append_event( + "run-1", + artifacts.RunEventDraft( + event_id="run-1-completed", + event_type="run.completed", + payload={}, + ), + ) def finalize(filename: str): barrier.wait() @@ -140,9 +159,27 @@ def test_concurrent_manifest_finalization_commits_one_authoritative_barrier( if event.event_type == "artifact.manifest.finalized" ] assert len(manifests) == 1 - assert len(events) == 2 + assert len(events) == 3 assert {result.event_id for result in results} == { manifests[0].event_id } finally: journal.close() + + +def test_discovery_marks_exact_result_cap_as_partial(tmp_path): + output_root = tmp_path / "output" + output_root.mkdir() + for name in ("a.txt", "b.txt"): + (output_root / name).write_text(name, encoding="utf-8") + snapshot = SimpleNamespace( + task_output_root=str(output_root), + working_directory=str(output_root), + task_start_time=0, + ) + + result = artifacts.discover_task_changed_files(snapshot, max_entries=1) + + assert len(result.artifacts) == 1 + assert result.scan_status == "partial" + assert result.truncated is True diff --git a/backend/tests/app/utils/server/test_sync_step_journal.py b/backend/tests/app/utils/server/test_sync_step_journal.py index 4684d37d5..7e5d07abf 100644 --- a/backend/tests/app/utils/server/test_sync_step_journal.py +++ b/backend/tests/app/utils/server/test_sync_step_journal.py @@ -163,42 +163,26 @@ async def test_end_finalizes_artifacts_before_assistant_result_and_terminal( monkeypatch, ): order: list[str] = [] - recorder = SimpleNamespace( - record_assistant_final=AsyncMock( - side_effect=lambda **_kwargs: order.append("assistant.final") - ) - ) - journal = SimpleNamespace( - get_run=lambda _run_id: SimpleNamespace( - run_id="run-1", project_id="project-1" - ) - ) + from app import run_runtime - from app import artifacts, run_runtime - from app.run_journal import runtime as journal_runtime + async def complete_turn(_run_id, *, project_id, assistant_data): + assert project_id == "project-1" + assert assistant_data == {"message": "done"} + order.extend( + [ + "artifact.manifest.finalized", + "assistant.final+run.completed", + ] + ) + return True - monkeypatch.setattr( - artifacts, - "finalize_run_artifacts", - lambda _journal, _run: order.append("artifact.manifest.finalized"), - ) - monkeypatch.setattr( - journal_runtime, "get_default_run_journal", lambda: journal - ) monkeypatch.setattr( run_runtime, "get_default_run_coordinator", lambda: SimpleNamespace( - complete_turn=AsyncMock( - side_effect=lambda _run_id: ( - order.append("run.completed") or True - ) - ) + complete_turn=AsyncMock(side_effect=complete_turn) ), ) - monkeypatch.setattr( - sync_step_module, "get_default_event_recorder", lambda: recorder - ) monkeypatch.setattr(sync_step_module, "env", lambda *_args: "") @sync_step_module.sync_step @@ -214,8 +198,7 @@ async def test_end_finalizes_artifacts_before_assistant_result_and_terminal( assert len(values) == 1 assert order == [ "artifact.manifest.finalized", - "assistant.final", - "run.completed", + "assistant.final+run.completed", "yielded", ] diff --git a/backend/tests/app/utils/test_workspace_resolver_direct_write.py b/backend/tests/app/utils/test_workspace_resolver_direct_write.py index e71a5e9c6..56cf03b23 100644 --- a/backend/tests/app/utils/test_workspace_resolver_direct_write.py +++ b/backend/tests/app/utils/test_workspace_resolver_direct_write.py @@ -31,12 +31,37 @@ from unittest.mock import patch import pytest from app.utils.workspace_resolver import ( + TaskSnapshot, WorkspaceBinding, WorkspaceResolver, WorkspaceStore, ) +def test_find_snapshot_resolves_run_owner_without_mutable_task_lock( + monkeypatch, tmp_path: Path +): + monkeypatch.setenv("HOME", str(tmp_path)) + store = WorkspaceStore() + snapshot = TaskSnapshot( + task_id="run-1", + project_id="project-1", + space_id="space-1", + user_id="42", + working_directory=str(tmp_path / "workspace"), + task_output_root=str(tmp_path / "output"), + task_start_time=1.0, + binding_source="default", + created_at="2026-08-14T00:00:00Z", + ) + store.save_snapshot("user@example.com", snapshot) + + located = store.find_snapshot("run-1") + + assert located == ("user_42", snapshot) + assert store.find_snapshot("../run-1") is None + + @pytest.fixture def bound_resolver(tmp_path: Path): """A resolver whose binding store returns a folder pointing at tmp_path. diff --git a/src/lib/projector/index.ts b/src/lib/projector/index.ts index b251d2a80..4ba3da083 100644 --- a/src/lib/projector/index.ts +++ b/src/lib/projector/index.ts @@ -1,3 +1,17 @@ +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + export * from './decode'; export * from './effects'; export * from './importers'; @@ -60,6 +74,15 @@ export function projectSnapshot( snapshot.recent_events, 'rehydrate' ).state; + const artifactProjection = projectRawEvents( + snapshot.project_id, + (snapshot.artifact_events || []).map((raw) => ({ + ...normalizeEvent(raw), + cloudCursor: null, + source: 'chat_step_v1' as const, + })), + 'rehydrate' + ).state; const runs = { ...projected.runs }; for (const aggregate of snapshot.runs || []) { const recent = runs[aggregate.run_id]; @@ -162,6 +185,16 @@ export function projectSnapshot( ? null : previous.resyncTargetCursor, runs, + artifactsByRun: mergeExistingState + ? { + ...(previous.artifactsByRun || {}), + ...projected.artifactsByRun, + ...artifactProjection.artifactsByRun, + } + : { + ...projected.artifactsByRun, + ...artifactProjection.artifactsByRun, + }, legacySteps, unknownEvents, }; diff --git a/src/lib/projector/reduce.ts b/src/lib/projector/reduce.ts index 0cb716e91..b58fb562f 100644 --- a/src/lib/projector/reduce.ts +++ b/src/lib/projector/reduce.ts @@ -1,8 +1,23 @@ +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + import type { CanonicalProjectEvent, - ProjectViewState, + ProjectedArtifact, ProjectedRun, ProjectorMode, + ProjectViewState, } from './types'; const RUN_STATUS_BY_EVENT: Record = { @@ -101,6 +116,7 @@ export function createProjectViewState( resyncReason: null, resyncTargetCursor: null, runs: {}, + artifactsByRun: {}, legacySteps: [], unknownEvents: [], }; @@ -208,6 +224,102 @@ export function reduceProjectView( (event.payload.__legacy_step_id as number | string | undefined) || event.eventId; const legacyData = event.payload.__legacy_data ?? event.payload; + let artifactsByRun = state.artifactsByRun; + if (event.eventType === 'artifact.manifest.finalized') { + const rawArtifacts = Array.isArray(event.payload.artifacts) + ? event.payload.artifacts + : []; + const projectedArtifacts: ProjectedArtifact[] = rawArtifacts.flatMap( + (raw) => { + if (!raw || typeof raw !== 'object') return []; + const value = raw as Record; + const relativePath = + typeof value.relativePath === 'string' ? value.relativePath : ''; + const name = + typeof value.filename === 'string' + ? value.filename + : relativePath.split('/').filter(Boolean).at(-1) || ''; + if (!relativePath || !name) return []; + return [ + { + artifactId: + typeof value.artifact_id === 'string' + ? value.artifact_id + : `${event.runId}:${relativePath}`, + runId: event.runId, + name, + relativePath, + changeType: + value.changeType === 'generated' ? 'generated' : 'changed', + size: + typeof value.size === 'number' && Number.isFinite(value.size) + ? value.size + : null, + modifiedAt: + typeof value.modifiedAt === 'number' && + Number.isFinite(value.modifiedAt) + ? value.modifiedAt + : null, + uploadPolicy: + typeof value.uploadPolicy === 'string' + ? value.uploadPolicy + : null, + localPathAvailable: value.localPathAvailable === true, + }, + ]; + } + ); + artifactsByRun = { + ...artifactsByRun, + [event.runId]: projectedArtifacts, + }; + } + if (event.eventType === 'artifact.uploaded') { + const artifactId = + typeof event.payload.artifact_id === 'string' + ? event.payload.artifact_id + : ''; + const rawAsset = + event.payload.asset_ref && typeof event.payload.asset_ref === 'object' + ? (event.payload.asset_ref as Record) + : null; + const key = typeof rawAsset?.key === 'string' ? rawAsset.key : ''; + if (artifactId && key) { + artifactsByRun = { + ...artifactsByRun, + [event.runId]: (artifactsByRun[event.runId] || []).map((artifact) => + artifact.artifactId === artifactId + ? { + ...artifact, + assetRef: { + key, + chatFileId: + typeof rawAsset?.chat_file_id === 'number' + ? rawAsset.chat_file_id + : undefined, + bucket: + typeof rawAsset?.bucket === 'string' + ? rawAsset.bucket + : undefined, + filename: + typeof rawAsset?.filename === 'string' + ? rawAsset.filename + : undefined, + size: + typeof rawAsset?.size === 'number' + ? rawAsset.size + : undefined, + contentType: + typeof rawAsset?.content_type === 'string' + ? rawAsset.content_type + : undefined, + }, + } + : artifact + ), + }; + } + } const hasLegacyStepId = event.legacyStep !== null && state.legacySteps.some( @@ -266,9 +378,12 @@ export function reduceProjectView( needsResync: state.needsResync, resyncReason: state.resyncReason, runs: { ...state.runs, [event.runId]: run }, + artifactsByRun, legacySteps, unknownEvents: - event.legacyStep || RUN_STATUS_BY_EVENT[event.eventType] + event.legacyStep || + RUN_STATUS_BY_EVENT[event.eventType] || + event.eventType.startsWith('artifact.') ? state.unknownEvents : [...state.unknownEvents, event], }; diff --git a/src/lib/projector/selectors.ts b/src/lib/projector/selectors.ts index c240b78a5..9844bcc02 100644 --- a/src/lib/projector/selectors.ts +++ b/src/lib/projector/selectors.ts @@ -1,7 +1,32 @@ -import type { ProjectedLegacyStep, ProjectViewState } from './types'; +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +import type { + ProjectedArtifact, + ProjectedLegacyStep, + ProjectViewState, +} from './types'; const CLOSED_ASK_STEPS = new Set(['end', 'human_reply']); +export function selectRunArtifacts( + view: ProjectViewState, + runId: string +): ProjectedArtifact[] { + return view.artifactsByRun[runId] || []; +} + export function selectPendingLegacyAsk( view: ProjectViewState, answeredStepIds: ReadonlySet diff --git a/src/lib/projector/types.ts b/src/lib/projector/types.ts index 4a3b3138b..d525ee333 100644 --- a/src/lib/projector/types.ts +++ b/src/lib/projector/types.ts @@ -1,3 +1,17 @@ +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + export type ProjectorMode = 'live' | 'rehydrate' | 'playback'; export type CanonicalProjectEvent = { @@ -37,6 +51,26 @@ export type ProjectedRun = { updatedAt: string; }; +export type ProjectedArtifact = { + artifactId: string; + runId: string; + name: string; + relativePath: string; + changeType: 'generated' | 'changed'; + size: number | null; + modifiedAt: number | null; + uploadPolicy: string | null; + localPathAvailable: boolean; + assetRef?: { + chatFileId?: number; + bucket?: string; + key: string; + filename?: string; + size?: number; + contentType?: string; + }; +}; + export type ProjectViewState = { projectId: string; mode: ProjectorMode; @@ -48,6 +82,7 @@ export type ProjectViewState = { resyncReason: string | null; resyncTargetCursor: number | null; runs: Record; + artifactsByRun: Record; legacySteps: ProjectedLegacyStep[]; unknownEvents: CanonicalProjectEvent[]; }; @@ -66,5 +101,6 @@ export type ProjectSnapshotInput = { updated_at: string; }>; recent_events: unknown[]; + artifact_events?: unknown[]; events_truncated?: boolean; }; diff --git a/src/lib/remoteControl.ts b/src/lib/remoteControl.ts index 454dbb0f1..79f8fc55f 100644 --- a/src/lib/remoteControl.ts +++ b/src/lib/remoteControl.ts @@ -316,6 +316,7 @@ export interface RemoteProjectSnapshot { updated_at: string; }>; recent_events: RemoteCanonicalEvent[]; + artifact_events?: RemoteCanonicalEvent[]; events_truncated: boolean; } diff --git a/src/store/chatStore.durableReplay.test.ts b/src/store/chatStore.durableReplay.test.ts index fe33654af..2d853c540 100644 --- a/src/store/chatStore.durableReplay.test.ts +++ b/src/store/chatStore.durableReplay.test.ts @@ -17,6 +17,7 @@ import { acceptCanonicalRunEvent, admitDurableRunResume, canonicalRunEventToLegacyMessage, + collectTaskUploadFiles, createCanonicalRunEventCursor, mergeFileInfoLists, normalizeTaskArtifactFileList, @@ -208,6 +209,60 @@ describe('canonical Run replay projection', () => { ).toHaveLength(1); }); + it('keeps Cloud-safe Artifact metadata when the local path is redacted', () => { + const files = normalizeTaskArtifactFileList([ + { + artifact_id: 'art-cloud-1', + filename: 'report.csv', + relativePath: 'reports/report.csv', + changeType: 'generated', + uploadPolicy: 'agent_generated', + localPathAvailable: false, + }, + ]); + + expect(files).toEqual([ + expect.objectContaining({ + artifactId: 'art-cloud-1', + name: 'report.csv', + path: '', + relativePath: 'reports/report.csv', + localPathAvailable: false, + }), + ]); + }); + + it('uploads only agent-generated canonical Artifacts', () => { + const candidates = collectTaskUploadFiles( + [], + [], + [], + [ + { + artifactId: 'art-generated', + name: 'generated.csv', + type: 'csv', + path: '/workspace/generated.csv', + uploadPolicy: 'agent_generated', + }, + { + artifactId: 'art-local', + name: 'private.csv', + type: 'csv', + path: '/workspace/private.csv', + uploadPolicy: 'metadata_only', + }, + ] + ); + + expect(candidates).toEqual([ + expect.objectContaining({ + artifactId: 'art-generated', + path: '/workspace/generated.csv', + }), + ]); + }); + it('matches URL-encoded stream paths and legacy x-prefixed paths', () => { expect( mergeFileInfoLists( diff --git a/src/store/chatStore.ts b/src/store/chatStore.ts index a544f65bb..877e69f16 100644 --- a/src/store/chatStore.ts +++ b/src/store/chatStore.ts @@ -146,6 +146,22 @@ export const canonicalRunEventToLegacyMessage = ( : undefined, } as AgentMessage; } + if (event.event_type === 'artifact.uploaded') { + const payload = + event.payload && typeof event.payload === 'object' + ? (event.payload as Record) + : null; + if (!payload || typeof payload.artifact_id !== 'string') return null; + return { + step: AgentStep.ARTIFACT_UPLOADED, + data: payload, + timestamp: + typeof event.created_at === 'number' && + Number.isFinite(event.created_at) + ? event.created_at + : undefined, + } as AgentMessage; + } if (typeof event.legacy_step !== 'string' || !event.legacy_step) { return null; } @@ -577,12 +593,14 @@ interface UploadCandidate { name: string; uploadName: string; source: UploadFileSource; + artifactId?: string; } interface UploadOutcome { success: boolean; fileName: string; source: UploadFileSource; + artifactId?: string; response?: unknown; error?: unknown; } @@ -776,12 +794,17 @@ export function collectTaskUploadFiles( // folder. Reading/referencing a local file is never upload consent. for (const file of taskOutputFiles) { if (!file?.path || !file?.name || file.isFolder) continue; + // A canonical manifest is also the consent boundary. Files from the + // selected workspace are metadata-only; only Eigent/agent generated + // outputs may leave the device automatically. + if (file.uploadPolicy === 'metadata_only') continue; if (!isReadableLocalPath(file.path)) continue; uploadCandidates.push({ path: file.path, name: file.name, relativePath: file.relativePath, source: 'project_output', + artifactId: file.artifactId, }); } @@ -836,6 +859,7 @@ async function uploadTaskFiles( success: false, fileName: file.name, source: file.source, + artifactId: file.artifactId, error: 'IPC renderer is unavailable', }); continue; @@ -846,6 +870,7 @@ async function uploadTaskFiles( success: false, fileName: file.name, source: file.source, + artifactId: file.artifactId, error: result.error || 'Failed to read file', }); continue; @@ -873,6 +898,7 @@ async function uploadTaskFiles( success: true, fileName: file.uploadName, source: file.source, + artifactId: file.artifactId, response: uploadResponse, }); } catch (error) { @@ -881,6 +907,7 @@ async function uploadTaskFiles( success: false, fileName: file.uploadName, source: file.source, + artifactId: file.artifactId, error, }); } @@ -1182,12 +1209,16 @@ export function extractFinalOutputFileList( } type TaskArtifactChange = { + artifact_id?: unknown; filename?: unknown; path?: unknown; relativePath?: unknown; changeType?: unknown; size?: unknown; modifiedAt?: unknown; + uploadPolicy?: unknown; + localPathAvailable?: unknown; + asset_ref?: unknown; }; /** Convert Brain's capability-protected local artifact index into preview cards. */ @@ -1209,9 +1240,19 @@ export function normalizeTaskArtifactFileList(value: unknown): FileInfo[] { typeof candidate.relativePath === 'string' ? normalizeOutputPath(candidate.relativePath) : undefined; + const artifactId = + typeof candidate.artifact_id === 'string' + ? candidate.artifact_id.trim() + : ''; const type = getFileTypeFromName(name); - const identity = (relativePath || path).toLowerCase(); - if (!path || !name || !identity || seen.has(identity)) continue; + const identity = (artifactId || relativePath || path).toLowerCase(); + if (!name || !identity || seen.has(identity)) continue; + + const asset = + candidate.asset_ref && typeof candidate.asset_ref === 'object' + ? (candidate.asset_ref as Record) + : null; + const assetKey = typeof asset?.key === 'string' ? asset.key : undefined; seen.add(identity); files.push({ @@ -1221,6 +1262,35 @@ export function normalizeTaskArtifactFileList(value: unknown): FileInfo[] { relativePath, icon: FileText, isRemote: false, + artifactId: artifactId || undefined, + uploadPolicy: + candidate.uploadPolicy === 'agent_generated' + ? 'agent_generated' + : candidate.uploadPolicy === 'metadata_only' + ? 'metadata_only' + : undefined, + localPathAvailable: + typeof candidate.localPathAvailable === 'boolean' + ? candidate.localPathAvailable + : Boolean(path), + assetRef: assetKey + ? { + chatFileId: + typeof asset?.chat_file_id === 'number' + ? asset.chat_file_id + : undefined, + key: assetKey, + bucket: + typeof asset?.bucket === 'string' ? asset.bucket : undefined, + filename: + typeof asset?.filename === 'string' ? asset.filename : undefined, + size: typeof asset?.size === 'number' ? asset.size : undefined, + contentType: + typeof asset?.content_type === 'string' + ? asset.content_type + : undefined, + } + : undefined, artifactChange: candidate.changeType === 'generated' ? 'generated' : 'changed', size: @@ -4068,6 +4138,57 @@ const chatStore = (initial?: Partial) => return; } + if (agentMessages.step === AgentStep.ARTIFACT_UPLOADED) { + const lockedTaskId = getCurrentTaskId(); + const lockedTask = getCurrentChatStore().tasks[lockedTaskId]; + if (!lockedTask) return; + const artifactId = agentMessages.data.artifact_id; + const rawAsset = agentMessages.data.asset_ref; + const assetKey = rawAsset?.key; + if ( + typeof artifactId !== 'string' || + !rawAsset || + typeof rawAsset !== 'object' || + typeof assetKey !== 'string' + ) { + return; + } + lockedTask.artifactManifestFiles = ( + lockedTask.artifactManifestFiles || [] + ).map((file) => + file.artifactId === artifactId + ? { + ...file, + assetRef: { + chatFileId: + typeof rawAsset.chat_file_id === 'number' + ? rawAsset.chat_file_id + : undefined, + key: assetKey, + bucket: + typeof rawAsset.bucket === 'string' + ? rawAsset.bucket + : undefined, + filename: + typeof rawAsset.filename === 'string' + ? rawAsset.filename + : undefined, + size: + typeof rawAsset.size === 'number' + ? rawAsset.size + : undefined, + contentType: + typeof rawAsset.content_type === 'string' + ? rawAsset.content_type + : undefined, + }, + } + : file + ); + setUpdateCount(); + return; + } + if (agentMessages.step === AgentStep.BUDGET_NOT_ENOUGH) { console.log('error', agentMessages.data); showCreditsToast(); @@ -4512,11 +4633,15 @@ const chatStore = (initial?: Partial) => uploadTargetId, user_id )) as CamelLogUploadFile[]) || []; - const taskOutputFiles = tasks[ + const legacyTaskOutputFiles = tasks[ currentTaskId ].taskAssigning.flatMap((agent) => agent.tasks.flatMap((task) => task.fileList || []) ); + const taskOutputFiles = + completedTask.artifactManifestFinalized === true + ? completedTask.artifactManifestFiles || [] + : legacyTaskOutputFiles; const filesToUpload = collectTaskUploadFiles( camelLogFiles, tasks[currentTaskId].messages, @@ -4543,6 +4668,41 @@ const chatStore = (initial?: Partial) => console.error('Failed to upload files:', failedUploads); } + for (const result of uploadResults) { + if ( + !result.success || + result.source !== 'project_output' || + !result.artifactId || + !result.response || + typeof result.response !== 'object' + ) { + continue; + } + const asset = result.response as Record< + string, + unknown + >; + try { + await fetchPost( + `/runs/${encodeURIComponent(currentTaskId)}/artifacts/${encodeURIComponent(result.artifactId)}/uploaded`, + { + chat_file_id: asset.id, + s3_bucket: asset.s3_bucket, + s3_key: asset.s3_key, + filename: asset.filename, + file_size: asset.file_size, + file_type: asset.file_type, + } + ); + } catch (error) { + console.error( + 'Uploaded Artifact asset could not be journaled:', + result.artifactId, + error + ); + } + } + const generatedSuccessCount = uploadResults.filter( (result) => result.success && result.source === 'project_output' diff --git a/src/types/chatbox.d.ts b/src/types/chatbox.d.ts index c3132c5bd..70bf772f8 100644 --- a/src/types/chatbox.d.ts +++ b/src/types/chatbox.d.ts @@ -41,6 +41,21 @@ declare global { mimeType?: string; supportsRanges?: boolean; preview?: FilePreviewPayload; + /** Stable identity from the canonical Artifact event stream. */ + artifactId?: string; + /** Only agent-generated outputs may be uploaded automatically. */ + uploadPolicy?: 'agent_generated' | 'metadata_only'; + /** False for Cloud-restored metadata whose local path was redacted. */ + localPathAvailable?: boolean; + /** Durable Cloud asset reference populated after upload succeeds. */ + assetRef?: { + chatFileId?: number; + bucket?: string; + key: string; + filename?: string; + size?: number; + contentType?: string; + }; } interface ProjectInfo { @@ -183,6 +198,8 @@ declare global { safety_class?: string; target_resources?: string[]; artifacts?: Array>; + artifact_id?: string; + asset_ref?: Record; artifact_count?: number; scan_status?: string; manifest_digest?: string; diff --git a/src/types/constants.ts b/src/types/constants.ts index a436dcfb4..2813ba0ae 100644 --- a/src/types/constants.ts +++ b/src/types/constants.ts @@ -33,6 +33,7 @@ export const AgentStep = { TERMINAL: 'terminal', WRITE_FILE: 'write_file', ARTIFACT_MANIFEST: 'artifact_manifest', + ARTIFACT_UPLOADED: 'artifact_uploaded', TODO_STATE: 'todo_state', BUDGET_NOT_ENOUGH: 'budget_not_enough', CONTEXT_TOO_LONG: 'context_too_long', diff --git a/test/unit/lib/projector.test.ts b/test/unit/lib/projector.test.ts index 1acad99fb..d83e56261 100644 --- a/test/unit/lib/projector.test.ts +++ b/test/unit/lib/projector.test.ts @@ -1,3 +1,17 @@ +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + import { completeProjectViewResync, createProjectViewState, @@ -10,6 +24,7 @@ import { projectSnapshot, reduceProjectView, selectPendingLegacyAsk, + selectRunArtifacts, } from '@/lib/projector'; import { describe, expect, it } from 'vitest'; @@ -41,6 +56,67 @@ describe('projector pipeline', () => { expect(first.currentCursor).toBe(1); }); + it('restores durable Artifacts outside the bounded snapshot event tail', () => { + const snapshot = projectSnapshot({ + project_id: 'project-1', + current_cursor: 100, + recent_events: [ + event({ + event_id: 'recent-100', + cloud_cursor: 100, + run_sequence: 100, + }), + ], + artifact_events: [ + event({ + event_id: 'manifest-1', + cloud_cursor: 1, + event_type: 'artifact.manifest.finalized', + legacy_step: null, + payload: { + artifacts: [ + { + artifact_id: 'artifact-1', + filename: 'report.csv', + relativePath: 'reports/report.csv', + changeType: 'generated', + uploadPolicy: 'agent_generated', + localPathAvailable: false, + }, + ], + }, + }), + event({ + event_id: 'upload-1', + cloud_cursor: 2, + run_sequence: 2, + event_type: 'artifact.uploaded', + legacy_step: null, + payload: { + artifact_id: 'artifact-1', + asset_ref: { + chat_file_id: 7, + bucket: 'assets', + key: 'reports/report.csv', + }, + }, + }), + ], + events_truncated: true, + }); + + expect(snapshot.needsResync).toBe(false); + expect(selectRunArtifacts(snapshot, 'run-1')).toEqual([ + expect.objectContaining({ + artifactId: 'artifact-1', + assetRef: expect.objectContaining({ + chatFileId: 7, + key: 'reports/report.csv', + }), + }), + ]); + }); + it('detects both Project cursor and Run sequence gaps', () => { const first = reduceProjectView( createProjectViewState('project-1', 'live'),