421 lines
15 KiB
Python
421 lines
15 KiB
Python
"""External executor coordination, notification outbox and release truth gate.
|
|
|
|
OpenClaw/Codex task declarations are not execution evidence. This module wraps
|
|
``SQLiteExecutionTruthRegistry`` with a durable notification outbox. A successful
|
|
release gate requires both terminal task evidence and acknowledged STARTED /
|
|
SUCCEEDED owner notifications when those events are required by the manifest.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import threading
|
|
import uuid
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
from .task_execution_registry import (
|
|
ExecutionTaskManifest,
|
|
ExecutionTruthRecord,
|
|
ExecutionTruthState,
|
|
SQLiteExecutionTruthRegistry,
|
|
utc_now,
|
|
)
|
|
|
|
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
|
|
|
|
def sha256_value(value: str, label: str) -> str:
|
|
normalized = str(value or "").strip().lower()
|
|
if not _SHA256.fullmatch(normalized):
|
|
raise ValueError(f"{label} must be a lowercase SHA256 hex digest")
|
|
return normalized
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NotificationRecord:
|
|
notification_id: str
|
|
task_id: str
|
|
event_type: str
|
|
target_ref: str
|
|
payload: Mapping[str, Any]
|
|
status: str
|
|
created_at: str
|
|
delivered_at: str | None = None
|
|
receipt_sha256: str | None = None
|
|
error: str | None = None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class SQLiteNotificationOutbox:
|
|
"""Durable owner-notification delivery and acknowledgement records."""
|
|
|
|
def __init__(self, path: str | Path) -> None:
|
|
self.path = Path(path).resolve()
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._lock = threading.RLock()
|
|
with self._connect() as connection:
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS external_task_notifications (
|
|
notification_id TEXT PRIMARY KEY,
|
|
task_id TEXT NOT NULL,
|
|
event_type TEXT NOT NULL,
|
|
target_ref TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
delivered_at TEXT,
|
|
receipt_sha256 TEXT,
|
|
error TEXT,
|
|
UNIQUE(task_id, event_type)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_external_notification_status
|
|
ON external_task_notifications(status, created_at);
|
|
"""
|
|
)
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
connection = sqlite3.connect(str(self.path), timeout=30.0)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("PRAGMA synchronous=FULL")
|
|
return connection
|
|
|
|
@staticmethod
|
|
def _decode(row: sqlite3.Row) -> NotificationRecord:
|
|
return NotificationRecord(
|
|
notification_id=row["notification_id"],
|
|
task_id=row["task_id"],
|
|
event_type=row["event_type"],
|
|
target_ref=row["target_ref"],
|
|
payload=json.loads(row["payload_json"]),
|
|
status=row["status"],
|
|
created_at=row["created_at"],
|
|
delivered_at=row["delivered_at"],
|
|
receipt_sha256=row["receipt_sha256"],
|
|
error=row["error"],
|
|
)
|
|
|
|
def enqueue(
|
|
self,
|
|
*,
|
|
task_id: str,
|
|
event_type: str,
|
|
target_ref: str,
|
|
payload: Mapping[str, Any],
|
|
) -> NotificationRecord:
|
|
now = utc_now()
|
|
record = NotificationRecord(
|
|
notification_id=f"notify_{uuid.uuid4().hex}",
|
|
task_id=str(task_id).strip(),
|
|
event_type=str(event_type).strip().upper(),
|
|
target_ref=str(target_ref).strip(),
|
|
payload=json.loads(json.dumps(dict(payload), ensure_ascii=False, default=str)),
|
|
status="PENDING",
|
|
created_at=now,
|
|
)
|
|
if not record.task_id or not record.event_type or not record.target_ref:
|
|
raise ValueError("task_id, event_type and target_ref are required")
|
|
with self._lock, self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
existing = connection.execute(
|
|
"SELECT * FROM external_task_notifications WHERE task_id = ? AND event_type = ?",
|
|
(record.task_id, record.event_type),
|
|
).fetchone()
|
|
if existing is not None:
|
|
connection.rollback()
|
|
return self._decode(existing)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO external_task_notifications
|
|
(notification_id, task_id, event_type, target_ref, payload_json,
|
|
status, created_at, delivered_at, receipt_sha256, error)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
|
|
""",
|
|
(
|
|
record.notification_id,
|
|
record.task_id,
|
|
record.event_type,
|
|
record.target_ref,
|
|
json.dumps(record.payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")),
|
|
record.status,
|
|
record.created_at,
|
|
),
|
|
)
|
|
connection.commit()
|
|
return record
|
|
|
|
def get(self, notification_id: str) -> NotificationRecord:
|
|
with self._lock, self._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT * FROM external_task_notifications WHERE notification_id = ?",
|
|
(str(notification_id).strip(),),
|
|
).fetchone()
|
|
if row is None:
|
|
raise KeyError("notification not found")
|
|
return self._decode(row)
|
|
|
|
def acknowledge(self, notification_id: str, *, receipt_sha256: str) -> NotificationRecord:
|
|
digest = sha256_value(receipt_sha256, "receipt_sha256")
|
|
with self._lock, self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
row = connection.execute(
|
|
"SELECT * FROM external_task_notifications WHERE notification_id = ?",
|
|
(str(notification_id).strip(),),
|
|
).fetchone()
|
|
if row is None:
|
|
connection.rollback()
|
|
raise KeyError("notification not found")
|
|
current = self._decode(row)
|
|
if current.status == "DELIVERED":
|
|
if current.receipt_sha256 != digest:
|
|
connection.rollback()
|
|
raise RuntimeError("notification already acknowledged with another receipt")
|
|
connection.rollback()
|
|
return current
|
|
delivered_at = utc_now()
|
|
connection.execute(
|
|
"""
|
|
UPDATE external_task_notifications
|
|
SET status = 'DELIVERED', delivered_at = ?, receipt_sha256 = ?, error = NULL
|
|
WHERE notification_id = ?
|
|
""",
|
|
(delivered_at, digest, current.notification_id),
|
|
)
|
|
connection.commit()
|
|
return self.get(current.notification_id)
|
|
|
|
def fail(self, notification_id: str, *, error: str) -> NotificationRecord:
|
|
message = str(error or "").strip()
|
|
if not message:
|
|
raise ValueError("notification error is required")
|
|
with self._lock, self._connect() as connection:
|
|
connection.execute(
|
|
"UPDATE external_task_notifications SET status = 'FAILED', error = ? WHERE notification_id = ?",
|
|
(message, str(notification_id).strip()),
|
|
)
|
|
connection.commit()
|
|
return self.get(notification_id)
|
|
|
|
def list(
|
|
self,
|
|
*,
|
|
task_id: str | None = None,
|
|
status: str | None = None,
|
|
) -> list[NotificationRecord]:
|
|
conditions: list[str] = []
|
|
values: list[str] = []
|
|
if task_id:
|
|
conditions.append("task_id = ?")
|
|
values.append(str(task_id).strip())
|
|
if status:
|
|
conditions.append("status = ?")
|
|
values.append(str(status).strip().upper())
|
|
where = f" WHERE {' AND '.join(conditions)}" if conditions else ""
|
|
with self._lock, self._connect() as connection:
|
|
rows = connection.execute(
|
|
f"SELECT * FROM external_task_notifications{where} ORDER BY created_at, notification_id",
|
|
tuple(values),
|
|
).fetchall()
|
|
return [self._decode(row) for row in rows]
|
|
|
|
|
|
class ExternalTaskCoordinator:
|
|
"""Single safe lifecycle used by external executors and release gates."""
|
|
|
|
def __init__(
|
|
self,
|
|
registry: SQLiteExecutionTruthRegistry,
|
|
notifications: SQLiteNotificationOutbox,
|
|
) -> None:
|
|
self.registry = registry
|
|
self.notifications = notifications
|
|
|
|
@classmethod
|
|
def from_file(cls, path: str | Path) -> "ExternalTaskCoordinator":
|
|
return cls(
|
|
SQLiteExecutionTruthRegistry(path),
|
|
SQLiteNotificationOutbox(path),
|
|
)
|
|
|
|
def declare(self, manifest: ExecutionTaskManifest) -> ExecutionTruthRecord:
|
|
return self.registry.declare(manifest)
|
|
|
|
def declare_file(self, path: str | Path) -> ExecutionTruthRecord:
|
|
value = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
if not isinstance(value, Mapping):
|
|
raise ValueError("external task manifest must be a JSON object")
|
|
return self.declare(ExecutionTaskManifest.from_dict(value))
|
|
|
|
def register(self, task_id: str, *, executor_run_id: str) -> ExecutionTruthRecord:
|
|
return self.registry.register(task_id, executor_run_id=executor_run_id)
|
|
|
|
def retry_not_started(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
executor_run_id: str,
|
|
) -> ExecutionTruthRecord:
|
|
return self.registry.retry_not_started(
|
|
task_id,
|
|
executor_run_id=executor_run_id,
|
|
)
|
|
|
|
def claim(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
executor_run_id: str,
|
|
executor_id: str,
|
|
) -> ExecutionTruthRecord:
|
|
return self.registry.claim(
|
|
task_id,
|
|
executor_run_id=executor_run_id,
|
|
executor_id=executor_id,
|
|
)
|
|
|
|
def _notify(self, record: ExecutionTruthRecord, event_type: str) -> NotificationRecord | None:
|
|
event = event_type.upper()
|
|
if event not in set(record.manifest.notification_events):
|
|
return None
|
|
return self.notifications.enqueue(
|
|
task_id=record.task_id,
|
|
event_type=event,
|
|
target_ref=record.manifest.owner_notification_ref,
|
|
payload={
|
|
"task_id": record.task_id,
|
|
"target_version": record.manifest.target_version,
|
|
"executor": record.manifest.executor,
|
|
"executor_id": record.executor_id,
|
|
"executor_run_id": record.executor_run_id,
|
|
"state": record.state.value,
|
|
"current_step": record.current_step,
|
|
"total_steps": len(record.manifest.steps),
|
|
"timestamp": utc_now(),
|
|
},
|
|
)
|
|
|
|
def start(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
executor_run_id: str,
|
|
executor_id: str,
|
|
) -> ExecutionTruthRecord:
|
|
record = self.registry.start(
|
|
task_id,
|
|
executor_run_id=executor_run_id,
|
|
executor_id=executor_id,
|
|
)
|
|
self._notify(record, "STARTED")
|
|
return record
|
|
|
|
def heartbeat(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
executor_run_id: str,
|
|
executor_id: str,
|
|
current_step: int,
|
|
) -> ExecutionTruthRecord:
|
|
return self.registry.heartbeat(
|
|
task_id,
|
|
executor_run_id=executor_run_id,
|
|
executor_id=executor_id,
|
|
current_step=current_step,
|
|
)
|
|
|
|
def succeed(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
executor_run_id: str,
|
|
executor_id: str,
|
|
evidence: Mapping[str, str],
|
|
) -> ExecutionTruthRecord:
|
|
normalized = {
|
|
str(key).strip(): sha256_value(value, f"evidence[{key}]")
|
|
for key, value in evidence.items()
|
|
}
|
|
record = self.registry.succeed(
|
|
task_id,
|
|
executor_run_id=executor_run_id,
|
|
executor_id=executor_id,
|
|
evidence=normalized,
|
|
)
|
|
self._notify(record, "SUCCEEDED")
|
|
return record
|
|
|
|
def fail(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
executor_run_id: str,
|
|
executor_id: str,
|
|
error: Mapping[str, Any],
|
|
evidence: Mapping[str, str] | None = None,
|
|
) -> ExecutionTruthRecord:
|
|
normalized = None
|
|
if evidence is not None:
|
|
normalized = {
|
|
str(key).strip(): sha256_value(value, f"evidence[{key}]")
|
|
for key, value in evidence.items()
|
|
}
|
|
record = self.registry.fail(
|
|
task_id,
|
|
executor_run_id=executor_run_id,
|
|
executor_id=executor_id,
|
|
error=error,
|
|
evidence=normalized,
|
|
)
|
|
self._notify(record, "FAILED")
|
|
return record
|
|
|
|
def gate(self, task_id: str) -> Mapping[str, Any]:
|
|
before = self.registry.get(task_id)
|
|
truth = self.registry.gate(task_id)
|
|
after = self.registry.get(task_id)
|
|
if after.state != before.state and after.state in {
|
|
ExecutionTruthState.NOT_STARTED,
|
|
ExecutionTruthState.STALLED,
|
|
}:
|
|
self._notify(after, after.state.value)
|
|
notifications = self.notifications.list(task_id=task_id)
|
|
delivered = {item.event_type for item in notifications if item.status == "DELIVERED"}
|
|
required_success_notifications = {
|
|
item
|
|
for item in ("STARTED", "SUCCEEDED")
|
|
if item in set(after.manifest.notification_events)
|
|
}
|
|
notification_complete = required_success_notifications.issubset(delivered)
|
|
passed = bool(truth["passed"] and notification_complete)
|
|
return {
|
|
**dict(truth),
|
|
"passed": passed,
|
|
"release_blocking": not passed,
|
|
"notification_complete": notification_complete,
|
|
"required_success_notifications": sorted(required_success_notifications),
|
|
"delivered_notifications": sorted(delivered),
|
|
"notifications": [item.as_dict() for item in notifications],
|
|
}
|
|
|
|
def status(self, task_id: str) -> Mapping[str, Any]:
|
|
gate = self.gate(task_id)
|
|
record = self.registry.get(task_id)
|
|
return {
|
|
"record": record.as_dict(),
|
|
"gate": gate,
|
|
"events": self.registry.events(task_id),
|
|
}
|
|
|
|
@staticmethod
|
|
def content_receipt(value: Mapping[str, Any]) -> str:
|
|
encoded = json.dumps(dict(value), ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|