mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-26 17:12:03 +00:00
630 lines
21 KiB
Python
630 lines
21 KiB
Python
# ========= 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. =========
|
|
|
|
"""Canonical per-Run Artifact discovery and durable event recording.
|
|
|
|
SQLite is the source of truth for a finalized Run's Artifact manifest. The
|
|
filesystem is consulted once, before ``run.completed``; replay and Cloud
|
|
projection consume the committed events instead of reconstructing history.
|
|
"""
|
|
|
|
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
|
|
from app.utils.workspace_paths import (
|
|
get_eigent_root,
|
|
runtime_owner_key,
|
|
sanitize_identity,
|
|
)
|
|
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"}
|
|
_AGENT_GENERATED_UPLOAD_POLICY = "agent_generated"
|
|
_METADATA_ONLY_UPLOAD_POLICY = "metadata_only"
|
|
_SPACE_INTERNAL_ARTIFACT_NAMES = {"todo.md"}
|
|
_SPACE_INTERNAL_ARTIFACT_ROOTS = {"terminal_logs"}
|
|
|
|
|
|
@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,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _ArtifactRoot:
|
|
path: Path
|
|
scan_all: bool
|
|
upload_policy: str
|
|
|
|
|
|
def _task_change_roots(
|
|
snapshot: TaskSnapshot,
|
|
*,
|
|
working_root_upload_policy: str = _METADATA_ONLY_UPLOAD_POLICY,
|
|
) -> list[_ArtifactRoot]:
|
|
"""Return the independently-scoped roots used by Artifact discovery.
|
|
|
|
``scan_all`` and ``upload_policy`` deliberately remain separate. A
|
|
Brain-owned Space root is still scanned only inside the Run's mutation
|
|
window, but files attributed to that Run may be uploaded. A user-bound
|
|
folder keeps the same bounded scan while remaining metadata-only.
|
|
"""
|
|
|
|
output_root = Path(snapshot.task_output_root).expanduser().resolve()
|
|
working_root = Path(snapshot.working_directory).expanduser().resolve()
|
|
roots: list[_ArtifactRoot] = []
|
|
if output_root.is_dir():
|
|
roots.append(
|
|
_ArtifactRoot(
|
|
path=output_root,
|
|
scan_all=True,
|
|
upload_policy=_AGENT_GENERATED_UPLOAD_POLICY,
|
|
)
|
|
)
|
|
if working_root != output_root and working_root.is_dir():
|
|
roots.append(
|
|
_ArtifactRoot(
|
|
path=working_root,
|
|
scan_all=False,
|
|
upload_policy=working_root_upload_policy,
|
|
)
|
|
)
|
|
return roots
|
|
|
|
|
|
def _brain_owned_space_root(
|
|
snapshot: TaskSnapshot,
|
|
*,
|
|
email: str,
|
|
user_id: str | int | None,
|
|
) -> Path:
|
|
safe_space_id = sanitize_identity(
|
|
getattr(snapshot, "space_id", None)
|
|
).removeprefix("space_")
|
|
return (
|
|
get_eigent_root()
|
|
/ runtime_owner_key(email, user_id)
|
|
/ f"space_{safe_space_id or 'scratch'}"
|
|
).resolve()
|
|
|
|
|
|
def _working_root_upload_policy(
|
|
snapshot: TaskSnapshot,
|
|
*,
|
|
email: str,
|
|
user_id: str | int | None,
|
|
) -> str:
|
|
"""Return the upload boundary for one frozen Space working root."""
|
|
|
|
try:
|
|
working_root = Path(snapshot.working_directory).expanduser().resolve()
|
|
managed_root = _brain_owned_space_root(
|
|
snapshot,
|
|
email=email,
|
|
user_id=user_id,
|
|
)
|
|
except (OSError, RuntimeError, ValueError):
|
|
return _METADATA_ONLY_UPLOAD_POLICY
|
|
if working_root == managed_root:
|
|
return _AGENT_GENERATED_UPLOAD_POLICY
|
|
return _METADATA_ONLY_UPLOAD_POLICY
|
|
|
|
|
|
def _artifact_upload_policy(root: _ArtifactRoot, relative_path: str) -> str:
|
|
"""Keep execution internals out of the project-output upload lane."""
|
|
|
|
path = Path(relative_path)
|
|
if path.name in _SPACE_INTERNAL_ARTIFACT_NAMES:
|
|
return _METADATA_ONLY_UPLOAD_POLICY
|
|
if path.parts and path.parts[0] in _SPACE_INTERNAL_ARTIFACT_ROOTS:
|
|
return _METADATA_ONLY_UPLOAD_POLICY
|
|
return root.upload_policy
|
|
|
|
|
|
def _git_run_changed_artifacts(
|
|
journal: SQLiteRunJournal,
|
|
run: RunRecord,
|
|
) -> ArtifactScanResult | None:
|
|
"""Project exact committed Run changes from Git into Artifact metadata."""
|
|
|
|
materialization = journal.get_run_git_materialization(run.run_id)
|
|
if (
|
|
materialization is None
|
|
or materialization.workspace_base_commit is None
|
|
or materialization.promoted_commit is None
|
|
or materialization.materialization_state
|
|
not in {"promoted", "archived"}
|
|
):
|
|
return None
|
|
repository = journal.get_git_repository(materialization.repository_id)
|
|
project = journal.get_project_git_state(run.project_id)
|
|
if repository is None or project is None:
|
|
return None
|
|
|
|
try:
|
|
from app.workspace_git.backend import GitBackend
|
|
|
|
git = GitBackend()
|
|
changes = git.changed_paths_between(
|
|
Path(repository.root_path),
|
|
base_commit=materialization.workspace_base_commit,
|
|
target_commit=materialization.promoted_commit,
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to derive authoritative Git Artifact changes",
|
|
extra={"run_id": run.run_id},
|
|
)
|
|
return None
|
|
|
|
visible_root = Path(repository.root_path).expanduser().resolve()
|
|
if project.pending_apply:
|
|
if project.worktree_path is None:
|
|
return None
|
|
visible_root = Path(project.worktree_path).expanduser().resolve()
|
|
values: list[dict[str, Any]] = []
|
|
truncated = len(changes) > MAX_ARTIFACTS_PER_RUN
|
|
artifact_root = _ArtifactRoot(
|
|
path=visible_root,
|
|
scan_all=False,
|
|
upload_policy=_AGENT_GENERATED_UPLOAD_POLICY,
|
|
)
|
|
for change in changes[:MAX_ARTIFACTS_PER_RUN]:
|
|
if change.status in {"D", "T"}:
|
|
continue
|
|
path = visible_root / change.relative_path
|
|
try:
|
|
if path.is_symlink() or not path.is_file():
|
|
continue
|
|
stat_result = path.stat()
|
|
except OSError:
|
|
continue
|
|
values.append(
|
|
{
|
|
"filename": path.name,
|
|
"path": str(path.resolve()),
|
|
"relativePath": change.relative_path,
|
|
"changeType": (
|
|
"generated" if change.status == "A" else "changed"
|
|
),
|
|
"size": stat_result.st_size,
|
|
"modifiedAt": stat_result.st_mtime * 1000,
|
|
"supportsRanges": True,
|
|
"uploadPolicy": _artifact_upload_policy(
|
|
artifact_root, change.relative_path
|
|
),
|
|
}
|
|
)
|
|
return ArtifactScanResult(
|
|
artifacts=values,
|
|
scan_status="partial" if truncated else "complete",
|
|
truncated=truncated,
|
|
)
|
|
|
|
|
|
def discover_task_changed_files(
|
|
snapshot: TaskSnapshot,
|
|
max_entries: int = MAX_ARTIFACTS_PER_RUN,
|
|
modification_windows: tuple[tuple[float, float | None], ...] | None = None,
|
|
*,
|
|
working_root_upload_policy: str = _METADATA_ONLY_UPLOAD_POLICY,
|
|
list_files_fn: Callable[..., list[str]] = list_files,
|
|
) -> 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 artifact_root in _task_change_roots(
|
|
snapshot,
|
|
working_root_upload_policy=working_root_upload_policy,
|
|
):
|
|
root = artifact_root.path
|
|
if remaining <= 0:
|
|
scan_limited = True
|
|
break
|
|
if artifact_root.scan_all:
|
|
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 = bounded_list_files(
|
|
root,
|
|
limit=remaining - len(paths),
|
|
modified_after=modified_after,
|
|
modified_before=modified_before,
|
|
)
|
|
for window_path in window_paths:
|
|
if window_path in window_seen:
|
|
continue
|
|
window_seen.add(window_path)
|
|
paths.append(window_path)
|
|
for absolute_path in paths:
|
|
try:
|
|
path = Path(absolute_path).resolve()
|
|
if not path.is_file():
|
|
continue
|
|
identity = str(path)
|
|
if identity in seen_paths:
|
|
continue
|
|
relative_path = path.relative_to(root).as_posix()
|
|
stat_result = path.stat()
|
|
except (OSError, ValueError):
|
|
# A tool may atomically replace or remove a file while the
|
|
# final scan runs. One vanished file cannot poison the Run.
|
|
continue
|
|
|
|
seen_paths.add(identity)
|
|
remaining -= 1
|
|
result.append(
|
|
{
|
|
"filename": path.name,
|
|
"path": identity,
|
|
"relativePath": relative_path,
|
|
"changeType": (
|
|
"generated" if artifact_root.scan_all else "changed"
|
|
),
|
|
"size": stat_result.st_size,
|
|
"modifiedAt": stat_result.st_mtime * 1000,
|
|
"supportsRanges": True,
|
|
"uploadPolicy": _artifact_upload_policy(
|
|
artifact_root, relative_path
|
|
),
|
|
}
|
|
)
|
|
if remaining <= 0:
|
|
break
|
|
|
|
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,
|
|
*,
|
|
working_root_upload_policy: str = _METADATA_ONLY_UPLOAD_POLICY,
|
|
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,
|
|
working_root_upload_policy=working_root_upload_policy,
|
|
list_files_fn=list_files_fn,
|
|
).artifacts
|
|
|
|
|
|
def task_modification_windows(
|
|
journal: SQLiteRunJournal,
|
|
run_id: str,
|
|
project_id: str,
|
|
) -> tuple[tuple[tuple[float, float | None], ...] | None, bool]:
|
|
"""Return filesystem mtime windows owned by this Run's attempts."""
|
|
|
|
run = journal.get_run(run_id)
|
|
if run is None or run.project_id != project_id:
|
|
return None, False
|
|
|
|
attempts = journal.list_run_attempts(run_id)
|
|
windows: list[tuple[float, float | None]] = []
|
|
for attempt in attempts:
|
|
end = attempt.ended_at
|
|
if end is None and run.status not in _ACTIVE_RUN_STATUSES:
|
|
end = run.updated_at
|
|
windows.append((attempt.started_at - 1.0, end))
|
|
|
|
if not windows:
|
|
end = (
|
|
run.updated_at if run.status not in _ACTIVE_RUN_STATUSES else None
|
|
)
|
|
windows.append((run.created_at - 1.0, end))
|
|
|
|
return tuple(windows), run.status in {"completed", "failed", "cancelled"}
|
|
|
|
|
|
def _artifact_projection(
|
|
*, run_id: str, artifact: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
relative_path = str(artifact.get("relativePath") or "")
|
|
change_type = str(artifact.get("changeType") or "changed")
|
|
artifact_id = (
|
|
"art_"
|
|
+ hashlib.sha256(
|
|
f"{run_id}\0{change_type}\0{relative_path}".encode()
|
|
).hexdigest()[:32]
|
|
)
|
|
return {"artifact_id": artifact_id, **dict(artifact)}
|
|
|
|
|
|
def record_artifact_manifest(
|
|
journal: SQLiteRunJournal,
|
|
*,
|
|
run_id: str,
|
|
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."""
|
|
|
|
projected = [
|
|
_artifact_projection(run_id=run_id, artifact=item)
|
|
for item in artifacts
|
|
]
|
|
drafts: list[RunEventDraft] = []
|
|
for artifact in projected:
|
|
event_type = (
|
|
"artifact.created"
|
|
if artifact.get("changeType") == "generated"
|
|
else "artifact.modified"
|
|
)
|
|
event_digest = _canonical_digest(
|
|
{
|
|
"run_id": run_id,
|
|
"event_type": event_type,
|
|
"artifact": artifact,
|
|
}
|
|
)
|
|
drafts.append(
|
|
RunEventDraft(
|
|
# Cloud canonical event ids are capped at 64 characters.
|
|
event_id=f"ae_{event_digest[:61]}",
|
|
event_type=event_type,
|
|
payload=artifact,
|
|
)
|
|
)
|
|
|
|
manifest_body = {
|
|
"artifacts": projected,
|
|
"artifact_count": len(projected),
|
|
"scan_status": scan_status,
|
|
"truncated": truncated,
|
|
}
|
|
manifest_digest = _canonical_digest(
|
|
{
|
|
**manifest_body,
|
|
# Absolute paths are machine-local transport data and should not
|
|
# decide whether a logical manifest is the same across retries.
|
|
"artifacts": [
|
|
{key: value for key, value in item.items() if key != "path"}
|
|
for item in projected
|
|
],
|
|
}
|
|
)
|
|
manifest_payload = {**manifest_body, "manifest_digest": manifest_digest}
|
|
# The durable id covers the local payload as stored. ``manifest_digest``
|
|
# remains path-independent for logical comparison and Cloud projection.
|
|
event_storage_digest = _canonical_digest(
|
|
{"run_id": run_id, "payload": manifest_payload}
|
|
)
|
|
drafts.append(
|
|
RunEventDraft(
|
|
event_id=f"am_{event_storage_digest[:61]}",
|
|
event_type="artifact.manifest.finalized",
|
|
payload=manifest_payload,
|
|
)
|
|
)
|
|
committed = journal.append_artifact_manifest_events(
|
|
run_id,
|
|
drafts,
|
|
expected_project_id=project_id,
|
|
)
|
|
return committed
|
|
|
|
|
|
def finalize_run_artifacts(
|
|
journal: SQLiteRunJournal,
|
|
run: RunRecord,
|
|
) -> 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 and current_run.status in {
|
|
"completed",
|
|
"failed",
|
|
"cancelled",
|
|
}:
|
|
return existing
|
|
|
|
email: str | None = None
|
|
user_id: str | int | None = None
|
|
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 = (
|
|
resolver.store.get_snapshot(email, run.run_id, user_id)
|
|
if email
|
|
else None
|
|
)
|
|
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,
|
|
project_id=run.project_id,
|
|
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 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
|
|
)
|
|
scan_result = discover_task_changed_files(
|
|
snapshot,
|
|
modification_windows=windows,
|
|
working_root_upload_policy=_working_root_upload_policy(
|
|
snapshot,
|
|
email=email,
|
|
user_id=user_id,
|
|
),
|
|
)
|
|
git_result = _git_run_changed_artifacts(journal, run)
|
|
if git_result is None:
|
|
artifacts = scan_result.artifacts
|
|
scan_status = scan_result.scan_status
|
|
truncated = scan_result.truncated
|
|
else:
|
|
# Git owns Project output attribution. Keep direct Space internals
|
|
# (todo/terminal logs) discovered outside the Git workspace, while
|
|
# exact committed changes win for matching relative paths.
|
|
git_by_path = {
|
|
item["relativePath"]: item for item in git_result.artifacts
|
|
}
|
|
direct_only = [
|
|
item
|
|
for item in scan_result.artifacts
|
|
if item["relativePath"] not in git_by_path
|
|
]
|
|
ordered = sorted(
|
|
git_by_path.values(), key=lambda item: item["relativePath"]
|
|
) + sorted(direct_only, key=lambda item: item["relativePath"])
|
|
artifacts = ordered[:MAX_ARTIFACTS_PER_RUN]
|
|
truncated = (
|
|
scan_result.truncated
|
|
or git_result.truncated
|
|
or len(ordered) > MAX_ARTIFACTS_PER_RUN
|
|
)
|
|
scan_status = "partial" if truncated else "complete"
|
|
|
|
manifest = record_artifact_manifest(
|
|
journal,
|
|
run_id=run.run_id,
|
|
project_id=run.project_id,
|
|
artifacts=artifacts,
|
|
scan_status=scan_status,
|
|
truncated=truncated,
|
|
)
|
|
try:
|
|
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.
|
|
logger.exception(
|
|
"Failed to cache finalized Artifact manifest",
|
|
extra={"run_id": run.run_id},
|
|
)
|
|
return manifest
|
|
|
|
|
|
def finalize_recoverable_run_artifacts(
|
|
journal: SQLiteRunJournal,
|
|
) -> tuple[str, ...]:
|
|
"""Finalize crash-surviving manifests before startup changes Run status."""
|
|
|
|
finalized: list[str] = []
|
|
for run in journal.list_recoverable_runs():
|
|
try:
|
|
finalize_run_artifacts(journal, run)
|
|
except Exception: # noqa: BLE001 - isolate one damaged workspace
|
|
logger.exception(
|
|
"Artifact recovery skipped one Run",
|
|
extra={"run_id": run.run_id},
|
|
)
|
|
continue
|
|
finalized.append(run.run_id)
|
|
return tuple(finalized)
|