""" Tests for PaaS Run Workspace — persistent run directories. Tests cover: 1. create_run_workspace() creates correct directory structure 2. save_run_artifacts() writes output/trace/cost files 3. Multiple runs preserve all history (no deletion) 4. Context carries workspace_path through fork/extend 5. Workspace not created when agent_dir is empty """ import json import os import shutil import tempfile import pytest from agentpaas.engine.sandbox import ( create_run_workspace, save_run_artifacts, build_change_manifest, ExecutionResult, ) from lambdagent.core import Context, TraceEntry @pytest.fixture def agent_dir(): """Create a temporary agent directory for testing.""" d = tempfile.mkdtemp(prefix="test_agent_") yield d shutil.rmtree(d, ignore_errors=True) # ============================================================ # 1. create_run_workspace() # ============================================================ class TestCreateRunWorkspace: def test_creates_directory_structure(self, agent_dir): """Creates workspace/run_YYYYMMDD_HHMMSS/ with subdirs""" ws = create_run_workspace(agent_dir, "run_abc123", "hello", {"type": "simple"}) assert os.path.isdir(ws) assert os.path.isdir(os.path.join(ws, "code")) assert os.path.isdir(os.path.join(ws, "results")) assert os.path.isdir(os.path.join(ws, "final")) def test_saves_input_json(self, agent_dir): """input.json contains run_id and input text""" ws = create_run_workspace(agent_dir, "run_abc123", "test input", {}) with open(os.path.join(ws, "input.json")) as f: data = json.load(f) assert data["run_id"] == "run_abc123" assert data["input"] == "test input" def test_saves_config_yml(self, agent_dir): """config.yml is a snapshot of the agent config""" config = {"type": "react", "systemPrompt": "You are helpful"} ws = create_run_workspace(agent_dir, "run_abc123", "input", config) assert os.path.isfile(os.path.join(ws, "config.yml")) def test_workspace_under_agent_dir(self, agent_dir): """Workspace is under {agent_dir}/workspace/""" ws = create_run_workspace(agent_dir, "run_abc123", "input", {}) assert ws.startswith(os.path.join(agent_dir, "workspace")) def test_run_dir_has_timestamp(self, agent_dir): """Run directory name matches run_YYYYMMDD_HHMMSS pattern""" ws = create_run_workspace(agent_dir, "run_abc123", "input", {}) dirname = os.path.basename(ws) assert dirname.startswith("run_20") # ============================================================ # 2. save_run_artifacts() # ============================================================ class TestSaveRunArtifacts: def test_saves_output_json(self, agent_dir): ws = create_run_workspace(agent_dir, "run_123", "input", {}) save_run_artifacts(ws, "final result", [], 1000, status="completed") with open(os.path.join(ws, "output.json")) as f: data = json.load(f) assert data["output"] == "final result" assert data["status"] == "completed" def test_saves_trace_json(self, agent_dir): ws = create_run_workspace(agent_dir, "run_123", "input", {}) trace = [TraceEntry("agent1", "id1", "in", "out", 100.0, "model", 50)] save_run_artifacts(ws, "result", trace, 500) with open(os.path.join(ws, "trace.json")) as f: data = json.load(f) assert len(data) == 1 assert data[0]["term_name"] == "agent1" assert data[0]["tokens_used"] == 50 def test_saves_cost_json(self, agent_dir): ws = create_run_workspace(agent_dir, "run_123", "input", {}) save_run_artifacts(ws, "result", [], 1500, input_tokens=100, steps=5) with open(os.path.join(ws, "cost.json")) as f: data = json.load(f) assert data["duration_ms"] == 1500 assert data["input_tokens"] == 100 assert data["steps"] == 5 def test_saves_on_failure(self, agent_dir): ws = create_run_workspace(agent_dir, "run_123", "input", {}) save_run_artifacts(ws, "", [], 500, error="boom", status="failed") with open(os.path.join(ws, "output.json")) as f: data = json.load(f) assert data["status"] == "failed" assert data["error"] == "boom" def test_noop_without_workspace(self): """save_run_artifacts does nothing if workspace_path is empty""" save_run_artifacts("", "result", [], 100) # Should not raise save_run_artifacts("/nonexistent/path", "result", [], 100) # Should not raise # ============================================================ # 3. Multiple Runs Preserve History # ============================================================ class TestMultipleRuns: def test_multiple_runs_coexist(self, agent_dir): """Each run creates a separate directory, old ones are not deleted""" import time ws1 = create_run_workspace(agent_dir, "run_aaa", "input1", {"v": 1}) time.sleep(1.1) # Ensure different timestamp ws2 = create_run_workspace(agent_dir, "run_bbb", "input2", {"v": 2}) assert ws1 != ws2 assert os.path.isdir(ws1) # First run still exists assert os.path.isdir(ws2) # Second run also exists # Both have their own input files with open(os.path.join(ws1, "input.json")) as f: assert json.load(f)["run_id"] == "run_aaa" with open(os.path.join(ws2, "input.json")) as f: assert json.load(f)["run_id"] == "run_bbb" def test_workspace_dir_lists_all_runs(self, agent_dir): """workspace/ directory contains all run directories""" import time create_run_workspace(agent_dir, "run_1", "i1", {}) time.sleep(1.1) create_run_workspace(agent_dir, "run_2", "i2", {}) ws_dir = os.path.join(agent_dir, "workspace") runs = os.listdir(ws_dir) assert len(runs) == 2 assert all(r.startswith("run_") for r in runs) # ============================================================ # 4. Context Carries workspace_path # ============================================================ class TestContextWorkspace: def test_context_workspace_path(self): ctx = Context(workspace_path="/tmp/test_ws", run_id="run_123") assert ctx.workspace_path == "/tmp/test_ws" assert ctx.run_id == "run_123" def test_fork_preserves_workspace(self): ctx = Context(workspace_path="/tmp/ws", run_id="run_abc") forked = ctx.fork() assert forked.workspace_path == "/tmp/ws" assert forked.run_id == "run_abc" def test_extend_preserves_workspace(self): ctx = Context(workspace_path="/tmp/ws", run_id="run_abc") child = ctx.extend(x=42) assert child.workspace_path == "/tmp/ws" assert child.run_id == "run_abc" def test_default_workspace_is_none(self): ctx = Context() assert ctx.workspace_path is None assert ctx.run_id is None # ============================================================ # 5. No Workspace When agent_dir Empty # ============================================================ class TestNoWorkspaceWithoutAgentDir: def test_execution_result_default_empty(self): r = ExecutionResult() assert r.workspace_path == "" # ============================================================ # 6. Change Manifest (修改清单) # ============================================================ class TestChangeManifest: def test_save_run_artifacts_writes_manifest(self, agent_dir): """save_run_artifacts auto-generates manifest.json""" ws = create_run_workspace(agent_dir, "run_m1", "input", {}) save_run_artifacts(ws, "result", [], 100) assert os.path.isfile(os.path.join(ws, "manifest.json")) def test_manifest_lists_agent_produced_files(self, agent_dir): """Files the agent wrote under the run dir appear in the manifest""" ws = create_run_workspace(agent_dir, "run_m2", "input", {}) with open(os.path.join(ws, "code", "main.py"), "w") as f: f.write("print('hi')\n") with open(os.path.join(ws, "results", "data.csv"), "w") as f: f.write("a,b\n1,2\n") manifest = build_change_manifest(ws) paths = {e["path"] for e in manifest["files"]} assert "code/main.py" in paths assert "results/data.csv" in paths assert manifest["file_count"] == 2 assert manifest["total_bytes"] > 0 def test_manifest_excludes_framework_metadata(self, agent_dir): """input/config/output/trace/cost/manifest themselves are not counted""" ws = create_run_workspace(agent_dir, "run_m3", "input", {}) save_run_artifacts(ws, "result", [], 100) # writes output/trace/cost + manifest with open(os.path.join(ws, "manifest.json")) as f: manifest = json.load(f) paths = {e["path"] for e in manifest["files"]} assert paths.isdisjoint({ "input.json", "config.yml", "output.json", "trace.json", "cost.json", "manifest.json", }) def test_manifest_records_sha256(self, agent_dir): """Each produced file carries a sha256 digest for the evidence chain""" ws = create_run_workspace(agent_dir, "run_m4", "input", {}) with open(os.path.join(ws, "final", "report.txt"), "w") as f: f.write("hello world") manifest = build_change_manifest(ws) entry = next(e for e in manifest["files"] if e["path"] == "final/report.txt") # sha256("hello world") assert entry["sha256"] == ( "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9" ) def test_manifest_carries_run_id(self, agent_dir): """run_id from input.json is propagated into the manifest""" ws = create_run_workspace(agent_dir, "run_m5", "input", {}) manifest = build_change_manifest(ws) assert manifest["run_id"] == "run_m5" def test_manifest_git_field_present(self, agent_dir): """git section is always present (available False outside a repo)""" ws = create_run_workspace(agent_dir, "run_m6", "input", {}) manifest = build_change_manifest(ws) assert "git" in manifest assert "available" in manifest["git"] def test_manifest_noop_on_invalid_path(self): """Invalid path returns {} and does not raise""" assert build_change_manifest("") == {} assert build_change_manifest("/nonexistent/xyz") == {} def test_empty_run_yields_empty_manifest(self, agent_dir): """A run that produced nothing still yields a valid (empty) manifest""" ws = create_run_workspace(agent_dir, "run_m7", "input", {}) manifest = build_change_manifest(ws) # only framework metadata present assert manifest["file_count"] == 0 assert manifest["files"] == []