File size: 3,092 Bytes
9a70a84 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """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())
|