vrr/eigent
2
0
Fork 0
mirror of https://github.com/eigent-ai/eigent.git synced 2026-08-31 02:05:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat: make artifacts durably replayable

Browse source
This commit is contained in:
4pmtong 2026-08-14 15:59:32 +08:00
parent 14bce81a84
commit ec6f081008
24 changed files with 1356 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

146
backend/app/artifacts.py
View file

@ -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.

93
backend/app/controller/run_controller.py
View file

@ -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()

97
backend/app/run_journal/store.py
View file

@ -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,

98
backend/app/run_runtime/coordinator.py
View file

@ -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,

73
backend/app/run_sync/cloud_sync.py
View file

@ -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]

38
backend/app/utils/file_utils.py
View file

@ -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

32
backend/app/utils/server/sync_step.py
View file

@ -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(

106
backend/app/utils/workspace_resolver.py
View file

@ -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:

69
backend/tests/app/controller/test_run_controller.py
View file

@ -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 (

15
backend/tests/app/run_runtime/test_coordinator.py
View file

@ -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")

65
backend/tests/app/run_sync/test_cloud_sync.py
View file

@ -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()

83
backend/tests/app/test_artifacts.py
View file

@ -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

43
backend/tests/app/utils/server/test_sync_step_journal.py
View file

@ -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",
]

25
backend/tests/app/utils/test_workspace_resolver_direct_write.py
View file

@ -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.

33
src/lib/projector/index.ts
View file

@ -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,
};

119
src/lib/projector/reduce.ts
View file

@ -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<string, ProjectedRun['status']> = {
@ -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<string, unknown>;
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<string, unknown>)
: 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],
};

27
src/lib/projector/selectors.ts
View file

@ -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<number | string>

36
src/lib/projector/types.ts
View file

@ -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<string, ProjectedRun>;
artifactsByRun: Record<string, ProjectedArtifact[]>;
legacySteps: ProjectedLegacyStep[];
unknownEvents: CanonicalProjectEvent[];
};
@ -66,5 +101,6 @@ export type ProjectSnapshotInput = {
updated_at: string;
}>;
recent_events: unknown[];
artifact_events?: unknown[];
events_truncated?: boolean;
};

1
src/lib/remoteControl.ts
View file

@ -316,6 +316,7 @@ export interface RemoteProjectSnapshot {
updated_at: string;
}>;
recent_events: RemoteCanonicalEvent[];
artifact_events?: RemoteCanonicalEvent[];
events_truncated: boolean;
}

55
src/store/chatStore.durableReplay.test.ts
View file

@ -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(

166
src/store/chatStore.ts
View file

@ -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<string, unknown>)
: 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<string, unknown>)
: 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<ChatStore>) =>
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<ChatStore>) =>
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<ChatStore>) =>
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'

17
src/types/chatbox.d.ts vendored
View file

@ -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<Record<string, unknown>>;
artifact_id?: string;
asset_ref?: Record<string, unknown>;
artifact_count?: number;
scan_status?: string;
manifest_digest?: string;

1
src/types/constants.ts
View file

@ -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',

76
test/unit/lib/projector.test.ts
View file

@ -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'),