[rollout] feat: add best-effort file artifacts for workflows/graders

docs/rollout-sdk.md

@@ -75,6 +75,28 @@ samples = await rollout_ctx.get_samples()            # async -> {name: RolloutSa
 
 `RolloutSample` ([types/sample.py](../osmosis_ai/rollout/types/sample.py)) fields: `id`, `messages`, `label`, `reward`, `remove_sample`, `metrics`, `extra_fields`.
 
+## Artifacts
+
+[../osmosis_ai/rollout/utils/file_artifacts.py](../osmosis_ai/rollout/utils/file_artifacts.py)
+
+Artifacts are files produced by a rollout — logs, traces, generated outputs, screenshots, or other large/binary data that does not belong in the sample.
+
+There is one rule: write files under `ctx.artifacts_dir` (available on both `AgentWorkflowContext` and `GraderContext`). It normally exists before your code runs, but is `Path | None` — it can be `None` when the host directory could not be created — so guard before use. In Harbor-backed rollouts it is the sandbox's `/logs/artifacts/`; `LocalBackend` provides an isolated per-rollout directory. If a file is produced elsewhere, copy it in (`shutil.copy2(path, ctx.artifacts_dir / "name")`).
+
+```python
+import json
+
+async def grade(self, ctx: GraderContext) -> Any:
+    if ctx.artifacts_dir:
+        (ctx.artifacts_dir / "trace.json").write_text(
+            json.dumps({"score_reason": "matched rubric"})
+        )
+    for sample_id in ctx.get_samples():
+        ctx.set_sample_reward(sample_id, 1.0)
+```
+
+After each rollout the artifacts land on the host under `~/.osmosis/<rollout_id>/artifacts/`, mirroring Harbor's collected-trial layout (user files sit at `.../artifacts/logs/artifacts/<file>`). Artifact handling is best-effort and never affects rewards or rollout status.
+
 ## Configs
 
 [../osmosis_ai/rollout/types/config.py](../osmosis_ai/rollout/types/config.py)

osmosis_ai/rollout/backend/harbor/agent_runner.py

@@ -14,6 +14,7 @@
 from typing import Any
 
 from osmosis_ai.rollout.context import AgentWorkflowContext, RolloutContext
+from osmosis_ai.rollout.utils.file_artifacts import HARBOR_ARTIFACTS_DIR
 from osmosis_ai.rollout.utils.imports import resolve_object
 
 AGENT_LOGS_DIR = Path("/logs/agent")
@@ -58,6 +59,7 @@ async def run_workflow(
         prompt=prompt,
         config=workflow_config,
         metadata=config.get("metadata"),
+        artifacts_dir=HARBOR_ARTIFACTS_DIR,
     )
 
     meta: dict[str, Any] = {"id": config.get("id", "")}

osmosis_ai/rollout/backend/harbor/backend.py

@@ -38,6 +38,7 @@
     RolloutSample,
     RolloutStatus,
 )
+from osmosis_ai.rollout.utils.file_artifacts import default_artifact_root
 from osmosis_ai.rollout.utils.imports import to_import_path
 from osmosis_ai.rollout.utils.rewards import validate_samples_have_rewards
 
@@ -142,6 +143,7 @@ def __init__(
         self.trials_dir: Path = (
             trials_dir if trials_dir != Path("trials") else self.root_dir / "trials"
         )
+        self.artifact_root: Path = default_artifact_root()
 
         self.pending: dict[str, PendingTrial] = {}
         self.prebuilt_image_tag: str | None = None
@@ -444,13 +446,45 @@ async def on_verification_start(self, event: TrialHookEvent) -> None:
         pending.workflow_complete_called = True
         await pending.on_workflow_complete(result)
 
+    def _relocate_trial_artifacts(self, rollout_id: str, *, move: bool) -> bool:
+        """Persist collected artifacts to the artifact root; ``move`` before
+        cleanup, else copy. Best-effort: returns ``False`` to keep the source."""
+        source_dir = self.trials_dir / f"{TRIAL_NAME_PREFIX}{rollout_id}" / "artifacts"
+        if not source_dir.is_dir():
+            return True
+        dest_dir = self.artifact_root / rollout_id / "artifacts"
+        try:
+            if dest_dir.exists():
+                shutil.rmtree(dest_dir)
+            dest_dir.parent.mkdir(parents=True, exist_ok=True)
+            if move:
+                shutil.move(source_dir, dest_dir)
+            else:
+                shutil.copytree(source_dir, dest_dir, symlinks=False)
+        except OSError:
+            logger.warning(
+                "Failed to relocate trial artifacts for rollout %s (best-effort)",
+                rollout_id,
+                exc_info=True,
+            )
+            return False
+        return True
+
     async def on_trial_end(self, event: TrialHookEvent) -> None:
         rollout_id = parse_rollout_id(event)
         pending = self.pending.pop(rollout_id, None)
         if not pending:
             logger.error("No pending trial found for rollout %s", rollout_id)
             return
 
+        # Evacuate artifacts first; delete the source only if it succeeds.
+        delete_trial = bool(
+            self.cleanup_successful_trials
+            and event.result
+            and not event.result.exception_info
+        )
+        relocated = self._relocate_trial_artifacts(rollout_id, move=delete_trial)
+
         if not pending.workflow_complete_called:
             if event.result and event.result.exception_info:
                 err = event.result.exception_info
@@ -518,9 +552,9 @@ async def on_trial_end(self, event: TrialHookEvent) -> None:
             rollout_dir = self.rollouts_dir / rollout_id
             shutil.rmtree(rollout_dir, ignore_errors=True)
 
-            if event.result and not event.result.exception_info:
-                trial_dir = self.trials_dir / f"{TRIAL_NAME_PREFIX}{rollout_id}"
-                shutil.rmtree(trial_dir, ignore_errors=True)
+        if delete_trial and relocated:
+            trial_dir = self.trials_dir / f"{TRIAL_NAME_PREFIX}{rollout_id}"
+            shutil.rmtree(trial_dir, ignore_errors=True)
 
         if not pending.done.done():
             pending.done.set_result(None)

osmosis_ai/rollout/backend/harbor/grader_runner.py

@@ -14,6 +14,7 @@
 
 from osmosis_ai.rollout.context import GraderContext
 from osmosis_ai.rollout.types import RolloutSample
+from osmosis_ai.rollout.utils.file_artifacts import HARBOR_ARTIFACTS_DIR
 from osmosis_ai.rollout.utils.imports import resolve_object
 
 VERIFIER_LOGS_DIR = Path("/logs/verifier")
@@ -71,7 +72,12 @@ def main() -> None:
         resolve_object(config["grader_config"]) if "grader_config" in config else None
     )
 
-    ctx = GraderContext(label=label, samples=samples, metadata=metadata)
+    ctx = GraderContext(
+        label=label,
+        samples=samples,
+        metadata=metadata,
+        artifacts_dir=HARBOR_ARTIFACTS_DIR,
+    )
     grader = grader_cls(grader_config)
     asyncio.run(grader.grade(ctx))
 

osmosis_ai/rollout/backend/local/backend.py

@@ -1,6 +1,7 @@
 import copy
 import logging
 import traceback
+from pathlib import Path
 from typing import Any
 
 from osmosis_ai.rollout.agent_workflow import AgentWorkflow
@@ -21,6 +22,10 @@
     RolloutStatus,
 )
 from osmosis_ai.rollout.utils.concurrency import ConcurrencyLimiter
+from osmosis_ai.rollout.utils.file_artifacts import (
+    HARBOR_COLLECTED_ARTIFACTS_SUBPATH,
+    default_artifact_root,
+)
 from osmosis_ai.rollout.utils.imports import resolve_object
 from osmosis_ai.rollout.utils.rewards import validate_samples_have_rewards
 
@@ -56,6 +61,8 @@ def __init__(
             max_concurrent=max_concurrent
         )
 
+        self.artifact_root: Path = default_artifact_root()
+
     @property
     def max_concurrency(self) -> int:
         return self.limiter.max_concurrent or 0
@@ -73,7 +80,9 @@ async def execute(
         on_grader_complete: ResultCallback | None = None,
     ) -> None:
         async with self.limiter.acquire():
-            result = await self.run_workflow(request)
+            artifacts_dir = self._make_artifacts_dir(request.id)
+
+            result = await self.run_workflow(request, artifacts_dir)
             await on_workflow_complete(result)
 
             if not on_grader_complete:
@@ -84,17 +93,39 @@ async def execute(
                 and (request.label is not None or request.metadata is not None)
                 and result.status == RolloutStatus.SUCCESS
             ):
-                graded = await self.run_grader(request, result)
+                graded = await self.run_grader(request, result, artifacts_dir)
                 await on_grader_complete(graded)
             else:
                 await on_grader_complete(ExecutionResult(status=RolloutStatus.FAILURE))
 
-    async def run_workflow(self, request: ExecutionRequest) -> ExecutionResult:
+    def _make_artifacts_dir(self, rollout_id: str) -> Path | None:
+        """Create the rollout's artifacts dir; ``None`` if it can't be created."""
+        artifacts_dir = (
+            self.artifact_root
+            / rollout_id
+            / "artifacts"
+            / HARBOR_COLLECTED_ARTIFACTS_SUBPATH
+        )
+        try:
+            artifacts_dir.mkdir(parents=True, exist_ok=True)
+        except OSError:
+            logger.warning(
+                "Failed to create artifacts dir for rollout %s (best-effort)",
+                rollout_id,
+                exc_info=True,
+            )
+            return None
+        return artifacts_dir
+
+    async def run_workflow(
+        self, request: ExecutionRequest, artifacts_dir: Path | None = None
+    ) -> ExecutionResult:
         config = copy.deepcopy(self.workflow_config)
         ctx = AgentWorkflowContext(
             prompt=request.prompt,
             config=config,
             metadata=request.metadata,
+            artifacts_dir=artifacts_dir,
         )
 
         rollout_ctx = get_rollout_context()
@@ -119,7 +150,10 @@ async def run_workflow(self, request: ExecutionRequest) -> ExecutionResult:
         )
 
     async def run_grader(
-        self, request: ExecutionRequest, result: ExecutionResult
+        self,
+        request: ExecutionRequest,
+        result: ExecutionResult,
+        artifacts_dir: Path | None = None,
     ) -> ExecutionResult:
         if not self.grader_cls:
             return result
@@ -128,6 +162,7 @@ async def run_grader(
             label=request.label,
             samples=result.samples,
             metadata=request.metadata,
+            artifacts_dir=artifacts_dir,
         )
         try:
             grader = self.grader_cls(self.grader_config)

osmosis_ai/rollout/context.py

@@ -2,6 +2,7 @@
 from abc import ABC, abstractmethod
 from contextvars import ContextVar
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import Any
 
 from osmosis_ai.rollout.types import (
@@ -97,6 +98,7 @@ class GraderContext:
     samples: dict[str, RolloutSample] = field(default_factory=dict)
     project_path: str | None = None
     metadata: dict[str, Any] | None = None
+    artifacts_dir: Path | None = None
 
     def get_samples(self) -> dict[str, RolloutSample]:
         return self.samples
@@ -112,16 +114,19 @@ class AgentWorkflowContext[TConfig: AgentWorkflowConfig]:
     prompt: list[dict[str, Any]]
     config: TConfig | None = None
     metadata: dict[str, Any] | None = None
+    artifacts_dir: Path | None = None
 
     def __init__(
         self,
         prompt: list[dict[str, Any]],
         config: TConfig | None = None,
         metadata: dict[str, Any] | None = None,
+        artifacts_dir: Path | None = None,
     ):
         self.prompt = prompt
         self.config = config
         self.metadata = metadata
+        self.artifacts_dir = artifacts_dir
 
 
 @dataclass
@@ -143,6 +148,12 @@ def __init__(
         config: TConfig,
         environment: Any = None,
         metadata: dict[str, Any] | None = None,
+        artifacts_dir: Path | None = None,
     ):
-        super().__init__(prompt=prompt, config=config, metadata=metadata)
+        super().__init__(
+            prompt=prompt,
+            config=config,
+            metadata=metadata,
+            artifacts_dir=artifacts_dir,
+        )
         self.environment = environment

osmosis_ai/rollout/utils/file_artifacts.py

@@ -0,0 +1,21 @@
+"""Rollout file artifacts: user code writes under ``ctx.artifacts_dir``."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+# Harbor's auto-collected convention directory inside the sandbox.
+HARBOR_ARTIFACTS_DIR = Path("/logs/artifacts")
+
+# Host subpath where Harbor lands it; LocalBackend reuses it to match.
+HARBOR_COLLECTED_ARTIFACTS_SUBPATH = HARBOR_ARTIFACTS_DIR.relative_to("/")
+
+
+def default_artifact_root() -> Path:
+    """Host directory that holds one subdirectory per rollout id.
+
+    Each rollout's outputs live at ``<root>/<rollout_id>/`` — file
+    artifacts under ``artifacts/`` and the archived trajectory as
+    ``trajectory.json`` — which the platform persists to durable storage.
+    """
+    return Path.home() / ".osmosis"

osmosis_ai/templates/_scaffolds/rollout/main.py.tpl

@@ -50,6 +50,11 @@ class MyGrader(Grader):
         # ctx.get_samples() returns {name: RolloutSample} for sources registered
         # by the workflow. Assign a scalar reward per sample with:
         #   ctx.set_sample_reward(name, value)
+        #
+        # Optional: write file artifacts (artifacts_dir may be None, so guard):
+        #   import json
+        #   if ctx.artifacts_dir:
+        #       (ctx.artifacts_dir / "debug.json").write_text(json.dumps({...}))
         for sample_id in ctx.get_samples():
             ctx.set_sample_reward(sample_id, 0.0)
 

tests/unit/rollout/test_context.py

@@ -1,5 +1,6 @@
 """Tests for osmosis_ai.rollout.context."""
 
+from pathlib import Path
 from unittest.mock import MagicMock
 
 import pytest
@@ -210,3 +211,13 @@ def test_metadata_threaded_through_super_init(self):
         )
         assert ctx.metadata == metadata
         assert ctx.environment is env
+
+    def test_artifacts_dir_threaded_through_super_init(self, tmp_path: Path):
+        cfg = AgentWorkflowConfig(name="harbor-test")
+        ctx = HarborAgentWorkflowContext(
+            prompt=[{"role": "user", "content": "hi"}],
+            config=cfg,
+            environment=MagicMock(),
+            artifacts_dir=tmp_path,
+        )
+        assert ctx.artifacts_dir == tmp_path