| """Detached worker for one durable terminal task.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import subprocess |
| import time |
| from pathlib import Path |
|
|
| from .artifacts import ArtifactStore |
| from .events import EventLog |
| from .sandbox import sandbox_argv, sandbox_environment |
| from .security import SecretRedactor |
| from .tasks import TaskStore |
|
|
|
|
| def run_terminal_task( |
| *, |
| task_root: str, |
| task_id: str, |
| workspace: str, |
| command: str, |
| ) -> int: |
| store = TaskStore(workspace) |
| expected_root = Path(task_root).expanduser().resolve() |
| if store.root != expected_root: |
| raise ValueError("task root does not match the selected workspace") |
| started = time.monotonic() |
| completed = subprocess.run( |
| sandbox_argv(workspace, command), |
| cwd=Path(workspace).expanduser().resolve(), |
| env=sandbox_environment(), |
| stdin=subprocess.DEVNULL, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| text=True, |
| check=False, |
| ) |
| current = store.get(task_id) |
| if current.status == "cancelled": |
| return 0 |
| redactor = SecretRedactor() |
| stdout = redactor.redact(completed.stdout) |
| stderr = redactor.redact(completed.stderr) |
| combined = stdout + (("\n" + stderr) if stderr else "") |
| artifact = ArtifactStore(workspace).put_text( |
| combined, |
| source=f"task:{task_id}", |
| session_id_sha256=current.session_id_sha256, |
| ) |
| result = { |
| "ok": completed.returncode == 0, |
| "exit_code": completed.returncode, |
| "stdout": stdout, |
| "stderr": stderr, |
| "elapsed_s": time.monotonic() - started, |
| "output_sha256": hashlib.sha256(combined.encode("utf-8")).hexdigest(), |
| "artifact_id": artifact.artifact_id, |
| } |
| updated = store.update( |
| task_id, |
| status="completed" if completed.returncode == 0 else "failed", |
| status_message="completed" if completed.returncode == 0 else "command failed", |
| progress={"finished": True}, |
| result=result, |
| error="" if completed.returncode == 0 else stderr, |
| process_id=0, |
| ) |
| EventLog(workspace).append( |
| "task_finished", |
| session_id_sha256=updated.session_id_sha256, |
| task_id=updated.task_id, |
| status=updated.status, |
| detail={ |
| "exit_code": completed.returncode, |
| "artifact_id": artifact.artifact_id, |
| "output_sha256": result["output_sha256"], |
| }, |
| ) |
| return completed.returncode |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description="Run one Nexum durable task") |
| parser.add_argument("--task-root", required=True) |
| parser.add_argument("--task-id", required=True) |
| parser.add_argument("--workspace", required=True) |
| parser.add_argument("--command", required=True) |
| args = parser.parse_args() |
| return run_terminal_task( |
| task_root=args.task_root, |
| task_id=args.task_id, |
| workspace=args.workspace, |
| command=args.command, |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
