502 lines
24 KiB
Python
502 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from contextlib import contextmanager
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sqlite3
|
|
import stat
|
|
import subprocess
|
|
import time
|
|
import threading
|
|
from typing import Any, Callable
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
|
|
|
|
OID_RE = re.compile(r"^[0-9a-f]{40}$")
|
|
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
|
|
TERMINAL = {"COMPLETED", "DEAD_LETTER", "REJECTED"}
|
|
CODEX_AUTH_LOCK = Path("/run/lock/aios-codex-auth.lock")
|
|
_CODEX_EXECUTION_LOCK = threading.Lock()
|
|
|
|
|
|
class WatcherError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@contextmanager
|
|
def _codex_auth_lock():
|
|
with _CODEX_EXECUTION_LOCK:
|
|
fd = os.open(CODEX_AUTH_LOCK, os.O_RDWR | getattr(os, "O_NOFOLLOW", 0))
|
|
try:
|
|
info = os.fstat(fd)
|
|
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) & ~0o660:
|
|
raise WatcherError("unsafe_codex_auth_lock")
|
|
fcntl.flock(fd, fcntl.LOCK_EX)
|
|
yield
|
|
finally:
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
os.close(fd)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Config:
|
|
active_dir: Path
|
|
completed_dir: Path
|
|
dead_letter_dir: Path
|
|
database_path: Path
|
|
audit_log_path: Path
|
|
gateway_endpoint: str
|
|
gateway_token_file: Path
|
|
contract_commit: str
|
|
lease_seconds: int
|
|
retry_seconds: int
|
|
max_attempts: int
|
|
poll_seconds: float
|
|
executor_path: Path | None
|
|
executor_args: tuple[str, ...]
|
|
executor_timeout_seconds: int
|
|
|
|
|
|
def _absolute(value: Any, key: str) -> Path:
|
|
if not isinstance(value, str) or not value or not Path(value).is_absolute():
|
|
raise WatcherError(f"invalid_{key}")
|
|
return Path(value)
|
|
|
|
|
|
def _secure_regular(path: Path, *, writable_mask: int = 0o022) -> os.stat_result:
|
|
info = os.lstat(path)
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
|
|
raise WatcherError(f"unsafe_file:{path}")
|
|
if stat.S_IMODE(info.st_mode) & writable_mask:
|
|
raise WatcherError(f"unsafe_mode:{path}")
|
|
return info
|
|
|
|
|
|
def _read_secure_text(
|
|
path: Path,
|
|
*,
|
|
writable_mask: int = 0o022,
|
|
max_bytes: int,
|
|
allowed_owners: set[int] | None = None,
|
|
) -> str:
|
|
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
|
fd = os.open(path, flags)
|
|
try:
|
|
info = os.fstat(fd)
|
|
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) & writable_mask:
|
|
raise WatcherError(f"unsafe_file:{path}")
|
|
if allowed_owners is not None and info.st_uid not in allowed_owners:
|
|
raise WatcherError(f"owner_mismatch:{path}")
|
|
if info.st_size > max_bytes:
|
|
raise WatcherError(f"file_too_large:{path}")
|
|
chunks: list[bytes] = []
|
|
total = 0
|
|
while True:
|
|
chunk = os.read(fd, min(65_536, max_bytes + 1 - total))
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
total += len(chunk)
|
|
if total > max_bytes:
|
|
raise WatcherError(f"file_too_large:{path}")
|
|
return b"".join(chunks).decode("utf-8")
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def load_config(path_value: str) -> Config:
|
|
path = _absolute(path_value, "config_path")
|
|
raw = json.loads(_read_secure_text(path, max_bytes=131_072, allowed_owners={0, os.geteuid()}))
|
|
expected = {
|
|
"active_dir", "completed_dir", "dead_letter_dir", "database_path",
|
|
"audit_log_path", "gateway_endpoint", "gateway_token_file",
|
|
"contract_commit", "lease_seconds", "retry_seconds", "max_attempts",
|
|
"poll_seconds", "executor_path", "executor_args",
|
|
"executor_timeout_seconds",
|
|
}
|
|
if not isinstance(raw, dict) or set(raw) != expected:
|
|
raise WatcherError("invalid_config_fields")
|
|
contract = raw["contract_commit"]
|
|
if not isinstance(contract, str) or not OID_RE.fullmatch(contract):
|
|
raise WatcherError("invalid_contract_commit")
|
|
endpoint = raw["gateway_endpoint"]
|
|
if not isinstance(endpoint, str) or not endpoint.startswith("http://127.0.0.1:") or not endpoint.endswith("/v1/orchestrator/events"):
|
|
raise WatcherError("gateway_endpoint_must_be_loopback")
|
|
executor_raw = raw["executor_path"]
|
|
executor = None if executor_raw is None else _absolute(executor_raw, "executor_path")
|
|
executor_args = raw["executor_args"]
|
|
if not isinstance(executor_args, list) or any(not isinstance(x, str) or not x for x in executor_args):
|
|
raise WatcherError("invalid_executor_args")
|
|
ints = ("lease_seconds", "retry_seconds", "max_attempts", "executor_timeout_seconds")
|
|
if any(not isinstance(raw[x], int) or raw[x] <= 0 for x in ints):
|
|
raise WatcherError("invalid_positive_integer")
|
|
if not isinstance(raw["poll_seconds"], (int, float)) or raw["poll_seconds"] <= 0:
|
|
raise WatcherError("invalid_poll_seconds")
|
|
return Config(
|
|
active_dir=_absolute(raw["active_dir"], "active_dir"),
|
|
completed_dir=_absolute(raw["completed_dir"], "completed_dir"),
|
|
dead_letter_dir=_absolute(raw["dead_letter_dir"], "dead_letter_dir"),
|
|
database_path=_absolute(raw["database_path"], "database_path"),
|
|
audit_log_path=_absolute(raw["audit_log_path"], "audit_log_path"),
|
|
gateway_endpoint=endpoint,
|
|
gateway_token_file=_absolute(raw["gateway_token_file"], "gateway_token_file"),
|
|
contract_commit=contract,
|
|
lease_seconds=raw["lease_seconds"], retry_seconds=raw["retry_seconds"],
|
|
max_attempts=raw["max_attempts"], poll_seconds=float(raw["poll_seconds"]),
|
|
executor_path=executor, executor_args=tuple(executor_args),
|
|
executor_timeout_seconds=raw["executor_timeout_seconds"],
|
|
)
|
|
|
|
|
|
def canonical_digest(value: dict[str, Any]) -> str:
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
def validate_command(value: Any, expected_contract: str) -> dict[str, Any]:
|
|
expected = {
|
|
"schema_version", "command_id", "project_key", "work_unit_id",
|
|
"contract_commit", "state_version", "execution",
|
|
}
|
|
if not isinstance(value, dict) or set(value) != expected or value.get("schema_version") != 1:
|
|
raise WatcherError("invalid_command_fields")
|
|
for key in ("command_id", "project_key", "work_unit_id"):
|
|
if not isinstance(value[key], str) or not ID_RE.fullmatch(value[key]):
|
|
raise WatcherError(f"invalid_{key}")
|
|
if value["contract_commit"] != expected_contract:
|
|
raise WatcherError("contract_commit_mismatch")
|
|
if not isinstance(value["state_version"], int) or value["state_version"] < 0:
|
|
raise WatcherError("invalid_state_version")
|
|
execution = value["execution"]
|
|
if not isinstance(execution, dict) or set(execution) != {"kind", "input"}:
|
|
raise WatcherError("invalid_execution")
|
|
if execution["kind"] not in {"acceptance_probe", "codex"}:
|
|
raise WatcherError("invalid_execution_kind")
|
|
if not isinstance(execution["input"], str) or len(execution["input"].encode()) > 65_536:
|
|
raise WatcherError("invalid_execution_input")
|
|
return value
|
|
|
|
|
|
class CommandWatcher:
|
|
def __init__(
|
|
self,
|
|
config: Config,
|
|
*,
|
|
clock: Callable[[], float] = time.time,
|
|
event_poster: Callable[[dict[str, Any]], None] | None = None,
|
|
executor: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
|
|
worker_id: str | None = None,
|
|
) -> None:
|
|
self.config = config
|
|
self.clock = clock
|
|
self.worker_id = worker_id or f"{os.uname().nodename}:{os.getpid()}:{uuid.uuid4().hex[:8]}"
|
|
self._uses_gateway = event_poster is None
|
|
self.event_poster = event_poster or self._post_event
|
|
self.executor = executor or self._execute
|
|
self._prepare()
|
|
|
|
def _prepare(self) -> None:
|
|
for directory in (self.config.active_dir, self.config.completed_dir, self.config.dead_letter_dir):
|
|
info = os.lstat(directory)
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) & 0o002:
|
|
raise WatcherError(f"unsafe_directory:{directory}")
|
|
self.config.database_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
self.config.audit_log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
with self._db() as db:
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
db.execute("""
|
|
CREATE TABLE IF NOT EXISTS commands (
|
|
command_id TEXT PRIMARY KEY, work_unit_id TEXT NOT NULL UNIQUE,
|
|
payload_sha256 TEXT NOT NULL,
|
|
source_path TEXT NOT NULL, status TEXT NOT NULL,
|
|
phase TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
|
|
lease_owner TEXT, lease_until REAL, next_attempt REAL NOT NULL DEFAULT 0,
|
|
result_json TEXT, start_event_json TEXT, end_event_json TEXT,
|
|
last_error TEXT, updated_at REAL NOT NULL
|
|
)
|
|
""")
|
|
|
|
def _db(self) -> sqlite3.Connection:
|
|
db = sqlite3.connect(self.config.database_path, timeout=5, isolation_level=None)
|
|
db.row_factory = sqlite3.Row
|
|
return db
|
|
|
|
def _audit(self, marker: str, command_id: str, **fields: Any) -> None:
|
|
safe = {k: v for k, v in fields.items() if k not in {"input", "token", "secret"}}
|
|
record = {"time": int(self.clock()), "marker": marker, "command_id": command_id, **safe}
|
|
fd = os.open(self.config.audit_log_path, os.O_APPEND | os.O_CREAT | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
|
|
try:
|
|
os.write(fd, (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode())
|
|
finally:
|
|
os.close(fd)
|
|
|
|
def _read_command(self, path: Path) -> tuple[dict[str, Any], str]:
|
|
raw = _read_secure_text(path, max_bytes=131_072, allowed_owners={0, os.geteuid()})
|
|
value = validate_command(json.loads(raw), self.config.contract_commit)
|
|
return value, canonical_digest(value)
|
|
|
|
def _register(self, command: dict[str, Any], digest: str, path: Path) -> str:
|
|
now = self.clock()
|
|
with self._db() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT payload_sha256,status FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
|
|
if row:
|
|
db.execute("COMMIT")
|
|
if row["payload_sha256"] != digest:
|
|
raise WatcherError("duplicate_command_payload_conflict")
|
|
return "duplicate_terminal" if row["status"] in TERMINAL else "existing"
|
|
prior_work_unit = db.execute(
|
|
"SELECT command_id FROM commands WHERE work_unit_id=?",
|
|
(command["work_unit_id"],),
|
|
).fetchone()
|
|
if prior_work_unit:
|
|
db.execute("COMMIT")
|
|
raise WatcherError("duplicate_work_unit")
|
|
db.execute(
|
|
"INSERT INTO commands(command_id,work_unit_id,payload_sha256,source_path,status,phase,updated_at) VALUES(?,?,?,?,?,?,?)",
|
|
(command["command_id"], command["work_unit_id"], digest, str(path), "PENDING", "START", now),
|
|
)
|
|
db.execute("COMMIT")
|
|
return "new"
|
|
|
|
def _claim(self, command_id: str) -> sqlite3.Row | None:
|
|
now = self.clock()
|
|
with self._db() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
|
|
if not row or row["status"] in TERMINAL or row["next_attempt"] > now:
|
|
db.execute("COMMIT")
|
|
return None
|
|
if row["lease_until"] is not None and row["lease_until"] > now and row["lease_owner"] != self.worker_id:
|
|
db.execute("COMMIT")
|
|
return None
|
|
attempts = row["attempts"] + 1
|
|
db.execute(
|
|
"UPDATE commands SET status='CLAIMED',attempts=?,lease_owner=?,lease_until=?,updated_at=? WHERE command_id=?",
|
|
(attempts, self.worker_id, now + self.config.lease_seconds, now, command_id),
|
|
)
|
|
row = db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
|
|
db.execute("COMMIT")
|
|
assert row is not None
|
|
self._audit("CODEX_AUTO_CLAIM", command_id, attempt=row["attempts"], worker=self.worker_id)
|
|
return row
|
|
|
|
def _update(self, command_id: str, *, status: str, phase: str, result: dict[str, Any] | None = None, error: str | None = None) -> None:
|
|
with self._db() as db:
|
|
release = status in TERMINAL
|
|
db.execute(
|
|
"UPDATE commands SET status=?,phase=?,result_json=COALESCE(?,result_json),last_error=?,lease_owner=CASE WHEN ? THEN NULL ELSE lease_owner END,lease_until=CASE WHEN ? THEN NULL ELSE lease_until END,updated_at=? WHERE command_id=?",
|
|
(status, phase, json.dumps(result, sort_keys=True) if result is not None else None, error, release, release, self.clock(), command_id),
|
|
)
|
|
|
|
def _retry_or_dead(self, command_id: str, path: Path, error: Exception) -> None:
|
|
message = f"{type(error).__name__}:{str(error)[:240]}"
|
|
with self._db() as db:
|
|
row = db.execute("SELECT attempts,phase FROM commands WHERE command_id=?", (command_id,)).fetchone()
|
|
attempts = row["attempts"] if row else self.config.max_attempts
|
|
if str(error) == "gateway_stale_state_version" and row:
|
|
column = "start_event_json" if row["phase"] == "START" else "end_event_json"
|
|
db.execute(f"UPDATE commands SET {column}=NULL WHERE command_id=?", (command_id,))
|
|
if attempts >= self.config.max_attempts:
|
|
db.execute("UPDATE commands SET status='DEAD_LETTER',last_error=?,lease_owner=NULL,lease_until=NULL,updated_at=? WHERE command_id=?", (message, self.clock(), command_id))
|
|
self._move(path, self.config.dead_letter_dir)
|
|
self._audit("DEAD_LETTER", command_id, error=message)
|
|
else:
|
|
db.execute("UPDATE commands SET status='RETRY',last_error=?,next_attempt=?,lease_owner=NULL,lease_until=NULL,updated_at=? WHERE command_id=?", (message, self.clock() + self.config.retry_seconds, self.clock(), command_id))
|
|
self._audit("RETRY_SCHEDULED", command_id, attempt=attempts, error=message)
|
|
|
|
def _event(self, command: dict[str, Any], event_type: str, status: str, suffix: str, result: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": 1, "project_key": command["project_key"],
|
|
"contract_commit": command["contract_commit"],
|
|
"event_id": f"{command['command_id']}:{suffix}", "source_agent": "codex",
|
|
"event_type": event_type, "work_unit_id": command["work_unit_id"],
|
|
"status": status, "state_version": command["state_version"],
|
|
"implementation_commit": None, "result_commit": None, "blocker": None,
|
|
"evidence_refs": [], "payload": {"command_id": command["command_id"], "result": result or {}},
|
|
}
|
|
|
|
def _gateway_token(self) -> str:
|
|
token = _read_secure_text(
|
|
self.config.gateway_token_file,
|
|
writable_mask=0o027,
|
|
max_bytes=16_384,
|
|
allowed_owners={0, os.geteuid()},
|
|
)
|
|
if not token or "\n" in token or "\r" in token:
|
|
raise WatcherError("invalid_gateway_token")
|
|
return token
|
|
|
|
def _gateway_state_version(self, project_key: str, fallback: int) -> int:
|
|
state_url = self.config.gateway_endpoint.removesuffix("/events") + "/state/" + urllib.parse.quote(project_key, safe="")
|
|
request = urllib.request.Request(
|
|
state_url,
|
|
headers={"Authorization": f"Bearer {self._gateway_token()}"},
|
|
method="GET",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
payload = json.loads(response.read(262_145))
|
|
version = payload.get("state_version")
|
|
if not isinstance(version, int) or version < 0:
|
|
raise WatcherError("invalid_gateway_state")
|
|
return version
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 404:
|
|
return fallback
|
|
raise WatcherError(f"gateway_state_http_{exc.code}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise WatcherError("gateway_state_unavailable") from exc
|
|
|
|
def _durable_event(self, command: dict[str, Any], column: str, event: dict[str, Any]) -> dict[str, Any]:
|
|
with self._db() as db:
|
|
row = db.execute(f"SELECT {column} FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
|
|
if row and row[column]:
|
|
return json.loads(row[column])
|
|
if self._uses_gateway:
|
|
event["state_version"] = self._gateway_state_version(command["project_key"], command["state_version"])
|
|
encoded = json.dumps(event, sort_keys=True, separators=(",", ":"))
|
|
with self._db() as db:
|
|
db.execute(f"UPDATE commands SET {column}=? WHERE command_id=? AND {column} IS NULL", (encoded, command["command_id"]))
|
|
saved = db.execute(f"SELECT {column} FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
|
|
if not saved or not saved[column]:
|
|
raise WatcherError("event_persistence_failed")
|
|
return json.loads(saved[column])
|
|
|
|
def _post_event(self, event: dict[str, Any]) -> None:
|
|
request = urllib.request.Request(
|
|
self.config.gateway_endpoint,
|
|
data=json.dumps(event, separators=(",", ":")).encode(),
|
|
headers={"Authorization": f"Bearer {self._gateway_token()}", "Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
if response.status not in {200, 201, 202}:
|
|
raise WatcherError(f"gateway_http_{response.status}")
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read(16_384)
|
|
try:
|
|
reason = json.loads(body).get("error")
|
|
except (json.JSONDecodeError, UnicodeDecodeError, AttributeError):
|
|
reason = None
|
|
if exc.code == 409 and reason == "stale_state_version":
|
|
raise WatcherError("gateway_stale_state_version") from exc
|
|
raise WatcherError(f"gateway_http_{exc.code}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise WatcherError("gateway_unavailable") from exc
|
|
|
|
def _execute(self, command: dict[str, Any]) -> dict[str, Any]:
|
|
execution = command["execution"]
|
|
if execution["kind"] == "acceptance_probe":
|
|
return {"kind": "acceptance_probe", "input_sha256": hashlib.sha256(execution["input"].encode()).hexdigest(), "exit_code": 0}
|
|
assert self.config.executor_path is not None
|
|
with _codex_auth_lock():
|
|
proc = subprocess.run(
|
|
[str(self.config.executor_path), *self.config.executor_args],
|
|
input=execution["input"], text=True, stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL, timeout=self.config.executor_timeout_seconds,
|
|
check=False, env={"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"},
|
|
)
|
|
if proc.returncode != 0:
|
|
raise WatcherError(f"codex_executor_failed:{proc.returncode}")
|
|
return {"kind": "codex", "exit_code": 0}
|
|
|
|
def _preflight_execution(self, command: dict[str, Any]) -> None:
|
|
if command["execution"]["kind"] == "acceptance_probe":
|
|
return
|
|
if self.config.executor_path is None:
|
|
raise WatcherError("codex_executor_unavailable")
|
|
info = _secure_regular(self.config.executor_path)
|
|
if not info.st_mode & stat.S_IXUSR:
|
|
raise WatcherError("codex_executor_not_executable")
|
|
|
|
def _move(self, source: Path, target_dir: Path) -> None:
|
|
if not source.exists() and not source.is_symlink():
|
|
return
|
|
target = target_dir / source.name
|
|
if target.exists() or target.is_symlink():
|
|
target = target_dir / f"{source.stem}-{uuid.uuid4().hex[:8]}{source.suffix}"
|
|
os.replace(source, target)
|
|
|
|
def _reject_unregistered(self, path: Path, error: Exception) -> None:
|
|
command_id = path.stem[:128] if ID_RE.fullmatch(path.stem[:128]) else f"invalid-{uuid.uuid4().hex[:12]}"
|
|
self._audit("COMMAND_REJECTED", command_id, error=f"{type(error).__name__}:{str(error)[:240]}")
|
|
self._move(path, self.config.dead_letter_dir)
|
|
|
|
def process_path(self, path: Path) -> bool:
|
|
try:
|
|
command, digest = self._read_command(path)
|
|
registration = self._register(command, digest, path)
|
|
if registration == "duplicate_terminal":
|
|
self._move(path, self.config.completed_dir)
|
|
return False
|
|
row = self._claim(command["command_id"])
|
|
if row is None:
|
|
return False
|
|
phase = row["phase"]
|
|
result = json.loads(row["result_json"]) if row["result_json"] else None
|
|
if phase == "START":
|
|
start_event = self._durable_event(
|
|
command,
|
|
"start_event_json",
|
|
self._event(command, "work_unit_started", "RUNNING", "started"),
|
|
)
|
|
self.event_poster(start_event)
|
|
self._audit("TASK_STARTED", command["command_id"])
|
|
self._update(command["command_id"], status="PENDING", phase="EXECUTE")
|
|
phase = "EXECUTE"
|
|
if phase == "EXECUTE":
|
|
self._preflight_execution(command)
|
|
self._update(command["command_id"], status="CLAIMED", phase="EXECUTING")
|
|
phase = "EXECUTING"
|
|
result = self.executor(command)
|
|
self._audit("TASK_EXECUTED", command["command_id"], result=result)
|
|
self._update(command["command_id"], status="PENDING", phase="END", result=result)
|
|
phase = "END"
|
|
elif phase == "EXECUTING":
|
|
raise WatcherError("ambiguous_execution_not_repeated")
|
|
if phase == "END":
|
|
assert result is not None
|
|
end_event = self._durable_event(
|
|
command,
|
|
"end_event_json",
|
|
self._event(command, "work_unit_ended", "PASS", "ended", result),
|
|
)
|
|
self.event_poster(end_event)
|
|
self._audit("TASK_ENDED", command["command_id"])
|
|
self._audit("EVENT_POSTED", command["command_id"], event="work_unit_ended")
|
|
self._update(command["command_id"], status="COMPLETED", phase="DONE", result=result)
|
|
self._move(path, self.config.completed_dir)
|
|
return True
|
|
except Exception as exc:
|
|
try:
|
|
if isinstance(exc, WatcherError) and str(exc) in {"duplicate_command_payload_conflict", "duplicate_work_unit"}:
|
|
self._reject_unregistered(path, exc)
|
|
elif "command" in locals():
|
|
self._retry_or_dead(command["command_id"], path, exc)
|
|
else:
|
|
self._reject_unregistered(path, exc)
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
def run_once(self) -> int:
|
|
failures = 0
|
|
for path in sorted(self.config.active_dir.glob("*.json")):
|
|
if not self.process_path(path):
|
|
failures += 1
|
|
return 1 if failures else 0
|
|
|
|
def run_forever(self) -> int:
|
|
while True:
|
|
self.run_once()
|
|
time.sleep(self.config.poll_seconds)
|