1090 lines
40 KiB
Python
1090 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from typing import Any
|
|
|
|
from .errors import ConflictError
|
|
from .models import Decision, Event
|
|
from .securefs import ensure_secure_directory
|
|
|
|
|
|
TRIGGER_ERROR_CATEGORIES = frozenset(
|
|
{"UNAUTHORIZED", "FORBIDDEN", "NOT_FOUND", "NOT_RUNNABLE", "NETWORK_ERROR"}
|
|
)
|
|
CALLBACK_STATES = frozenset({
|
|
"PENDING", "SENT", "AWAITING_CALLBACK", "CALLBACK_RECEIVED",
|
|
"APPLIED", "STALE_SUPERSEDED", "DEAD_LETTER",
|
|
})
|
|
|
|
|
|
class Store:
|
|
"""Durable state machine with idempotency, optimistic versioning, and leases."""
|
|
|
|
def __init__(self, path: Path):
|
|
self.path = Path(path)
|
|
ensure_secure_directory(self.path.parent)
|
|
self._local = threading.local()
|
|
self._initialize()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(self.path, timeout=30, isolation_level=None)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA synchronous=FULL")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
conn.execute("PRAGMA busy_timeout=30000")
|
|
return conn
|
|
|
|
def _conn(self) -> sqlite3.Connection:
|
|
conn = getattr(self._local, "conn", None)
|
|
if conn is None:
|
|
conn = self._connect()
|
|
self._local.conn = conn
|
|
return conn
|
|
|
|
def _initialize(self) -> None:
|
|
conn = self._connect()
|
|
try:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS project_state (
|
|
project_key TEXT PRIMARY KEY,
|
|
contract_commit TEXT NOT NULL,
|
|
state_version INTEGER NOT NULL,
|
|
state_json TEXT NOT NULL,
|
|
hops INTEGER NOT NULL DEFAULT 0,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS gate_hops (
|
|
project_key TEXT NOT NULL,
|
|
gate_id TEXT NOT NULL,
|
|
hops INTEGER NOT NULL DEFAULT 0,
|
|
updated_at INTEGER NOT NULL,
|
|
PRIMARY KEY(project_key, gate_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS events (
|
|
event_id TEXT PRIMARY KEY,
|
|
project_key TEXT NOT NULL,
|
|
digest TEXT NOT NULL,
|
|
event_json TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
channel TEXT,
|
|
remote_id TEXT,
|
|
trigger_http_status INTEGER,
|
|
trigger_error_category TEXT,
|
|
callback_state TEXT NOT NULL DEFAULT 'PENDING',
|
|
callback_deadline INTEGER,
|
|
callback_attempts INTEGER NOT NULL DEFAULT 0,
|
|
callback_received_at INTEGER,
|
|
applied_at INTEGER,
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
|
next_attempt_at INTEGER NOT NULL DEFAULT 0,
|
|
lease_owner TEXT,
|
|
lease_until INTEGER,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_events_poll
|
|
ON events(status, next_attempt_at, lease_until, created_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS decisions (
|
|
event_id TEXT PRIMARY KEY REFERENCES events(event_id),
|
|
project_key TEXT NOT NULL,
|
|
decision_json TEXT NOT NULL,
|
|
state_version_after INTEGER NOT NULL,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS outbox (
|
|
work_unit_id TEXT PRIMARY KEY,
|
|
event_id TEXT NOT NULL REFERENCES events(event_id),
|
|
assigned_agent TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
|
next_attempt_at INTEGER NOT NULL DEFAULT 0,
|
|
lease_owner TEXT,
|
|
lease_until INTEGER,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_outbox_poll
|
|
ON outbox(status, next_attempt_at, lease_until, created_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS lifecycle (
|
|
sync_id TEXT PRIMARY KEY,
|
|
project_key TEXT NOT NULL,
|
|
work_unit_id TEXT,
|
|
kind TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
|
next_attempt_at INTEGER NOT NULL DEFAULT 0,
|
|
lease_owner TEXT,
|
|
lease_until INTEGER,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_lifecycle_poll
|
|
ON lifecycle(status, next_attempt_at, lease_until, created_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS blocker_history (
|
|
project_key TEXT NOT NULL,
|
|
gate_id TEXT NOT NULL,
|
|
blocker TEXT NOT NULL,
|
|
repetitions INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
PRIMARY KEY(project_key, gate_id, blocker)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS summaries (
|
|
project_key TEXT PRIMARY KEY,
|
|
last_emitted_at INTEGER NOT NULL DEFAULT 0,
|
|
last_digest TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS observed_refs (
|
|
refname TEXT PRIMARY KEY,
|
|
object_id TEXT NOT NULL,
|
|
observed_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS metadata (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
columns = {
|
|
str(row["name"])
|
|
for row in conn.execute("PRAGMA table_info(events)").fetchall()
|
|
}
|
|
if "trigger_http_status" not in columns:
|
|
conn.execute("ALTER TABLE events ADD COLUMN trigger_http_status INTEGER")
|
|
if "trigger_error_category" not in columns:
|
|
conn.execute("ALTER TABLE events ADD COLUMN trigger_error_category TEXT")
|
|
migrations = {
|
|
"callback_state": "TEXT NOT NULL DEFAULT 'PENDING'",
|
|
"callback_deadline": "INTEGER",
|
|
"callback_attempts": "INTEGER NOT NULL DEFAULT 0",
|
|
"callback_received_at": "INTEGER",
|
|
"applied_at": "INTEGER",
|
|
}
|
|
for name, declaration in migrations.items():
|
|
if name not in columns:
|
|
conn.execute(f"ALTER TABLE events ADD COLUMN {name} {declaration}")
|
|
finally:
|
|
conn.close()
|
|
|
|
def register_event(self, event: Event) -> tuple[bool, dict[str, Any]]:
|
|
now = int(time.time())
|
|
conn = self._conn()
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
existing = conn.execute(
|
|
"SELECT digest, event_json FROM events WHERE event_id=?",
|
|
(event.event_id,),
|
|
).fetchone()
|
|
if existing:
|
|
if existing["digest"] != event.digest():
|
|
raise ConflictError("event_id_payload_conflict")
|
|
conn.execute("COMMIT")
|
|
return False, json.loads(existing["event_json"])
|
|
|
|
state = conn.execute(
|
|
"SELECT contract_commit, state_version FROM project_state WHERE project_key=?",
|
|
(event.project_key,),
|
|
).fetchone()
|
|
if state is None:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO project_state(
|
|
project_key, contract_commit, state_version, state_json,
|
|
hops, updated_at
|
|
) VALUES (?, ?, ?, '{}', 0, ?)
|
|
""",
|
|
(event.project_key, event.contract_commit, event.state_version, now),
|
|
)
|
|
else:
|
|
if state["contract_commit"] != event.contract_commit:
|
|
raise ConflictError("contract_version_mismatch")
|
|
if state["state_version"] != event.state_version:
|
|
raise ConflictError("stale_state_version")
|
|
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO events(
|
|
event_id, project_key, digest, event_json, status,
|
|
next_attempt_at, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, 'RECEIVED', ?, ?, ?)
|
|
""",
|
|
(
|
|
event.event_id,
|
|
event.project_key,
|
|
event.digest(),
|
|
event.canonical_json(),
|
|
now,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.execute("COMMIT")
|
|
return True, event.to_dict()
|
|
except Exception:
|
|
conn.execute("ROLLBACK")
|
|
raise
|
|
|
|
def claim_events(
|
|
self,
|
|
*,
|
|
owner: str,
|
|
statuses: tuple[str, ...],
|
|
limit: int,
|
|
lease_seconds: int,
|
|
) -> list[dict[str, Any]]:
|
|
if not statuses:
|
|
return []
|
|
now = int(time.time())
|
|
placeholders = ",".join("?" for _ in statuses)
|
|
conn = self._conn()
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT event_id FROM events
|
|
WHERE status IN ({placeholders})
|
|
AND next_attempt_at <= ?
|
|
AND (lease_until IS NULL OR lease_until < ?)
|
|
ORDER BY
|
|
CASE json_extract(event_json, '$.event_type')
|
|
WHEN 'central_gpt_health_probe' THEN 0
|
|
WHEN 'a0_unattended_acceptance_requested' THEN 0
|
|
ELSE 1
|
|
END,
|
|
CASE WHEN json_extract(event_json, '$.event_type') IN (
|
|
'central_gpt_health_probe', 'a0_unattended_acceptance_requested'
|
|
) THEN created_at END DESC,
|
|
created_at ASC
|
|
LIMIT ?
|
|
""",
|
|
(*statuses, now, now, limit),
|
|
).fetchall()
|
|
ids = [row["event_id"] for row in rows]
|
|
for event_id in ids:
|
|
conn.execute(
|
|
"""
|
|
UPDATE events SET lease_owner=?, lease_until=?, updated_at=?
|
|
WHERE event_id=?
|
|
""",
|
|
(owner, now + lease_seconds, now, event_id),
|
|
)
|
|
conn.execute("COMMIT")
|
|
except Exception:
|
|
conn.execute("ROLLBACK")
|
|
raise
|
|
if not ids:
|
|
return []
|
|
placeholders = ",".join("?" for _ in ids)
|
|
claimed = {
|
|
str(row["event_id"]): dict(row)
|
|
for row in conn.execute(
|
|
f"SELECT * FROM events WHERE event_id IN ({placeholders})",
|
|
ids,
|
|
).fetchall()
|
|
}
|
|
return [claimed[str(event_id)] for event_id in ids]
|
|
|
|
def update_event_trigger(
|
|
self,
|
|
event_id: str,
|
|
*,
|
|
status: str,
|
|
channel: str | None,
|
|
remote_id: str | None,
|
|
retry_after: int = 0,
|
|
increment_attempt: bool = False,
|
|
trigger_http_status: int | None = None,
|
|
trigger_error_category: str | None = None,
|
|
callback_state: str | None = None,
|
|
callback_deadline_after: int | None = None,
|
|
increment_callback_attempt: bool = False,
|
|
) -> None:
|
|
if trigger_http_status is not None and (
|
|
not isinstance(trigger_http_status, int)
|
|
or not 100 <= trigger_http_status <= 599
|
|
):
|
|
raise ValueError("invalid_trigger_http_status")
|
|
if (
|
|
trigger_error_category is not None
|
|
and trigger_error_category not in TRIGGER_ERROR_CATEGORIES
|
|
):
|
|
raise ValueError("invalid_trigger_error_category")
|
|
if callback_state is not None and callback_state not in CALLBACK_STATES:
|
|
raise ValueError("invalid_callback_state")
|
|
if callback_deadline_after is not None and callback_deadline_after < 0:
|
|
raise ValueError("invalid_callback_deadline")
|
|
now = int(time.time())
|
|
self._conn().execute(
|
|
"""
|
|
UPDATE events
|
|
SET status=?, channel=?, remote_id=?,
|
|
trigger_http_status=?, trigger_error_category=?,
|
|
callback_state=COALESCE(?, callback_state),
|
|
callback_deadline=?,
|
|
callback_attempts=callback_attempts + ?,
|
|
attempts=attempts + ?,
|
|
next_attempt_at=?,
|
|
lease_owner=NULL, lease_until=NULL, updated_at=?
|
|
WHERE event_id=?
|
|
""",
|
|
(
|
|
status,
|
|
channel,
|
|
remote_id,
|
|
trigger_http_status,
|
|
trigger_error_category,
|
|
callback_state,
|
|
None if callback_deadline_after is None else now + callback_deadline_after,
|
|
1 if increment_callback_attempt else 0,
|
|
1 if increment_attempt else 0,
|
|
now + max(0, retry_after),
|
|
now,
|
|
event_id,
|
|
),
|
|
)
|
|
|
|
def claim_workspace_timeouts(
|
|
self,
|
|
*,
|
|
owner: str,
|
|
older_than: int,
|
|
limit: int,
|
|
lease_seconds: int,
|
|
) -> list[dict[str, Any]]:
|
|
return self._claim_event_subset(
|
|
owner=owner,
|
|
where_sql="status='AWAITING_DECISION' AND channel='workspace_agent' AND callback_state='AWAITING_CALLBACK' AND callback_deadline <= ?",
|
|
where_params=(older_than,),
|
|
limit=limit,
|
|
lease_seconds=lease_seconds,
|
|
)
|
|
|
|
def claim_response_polls(
|
|
self,
|
|
*,
|
|
owner: str,
|
|
limit: int,
|
|
lease_seconds: int,
|
|
) -> list[dict[str, Any]]:
|
|
now = int(time.time())
|
|
return self._claim_event_subset(
|
|
owner=owner,
|
|
where_sql="status='POLLING_RESPONSE' AND next_attempt_at <= ?",
|
|
where_params=(now,),
|
|
limit=limit,
|
|
lease_seconds=lease_seconds,
|
|
)
|
|
|
|
def _claim_event_subset(
|
|
self,
|
|
*,
|
|
owner: str,
|
|
where_sql: str,
|
|
where_params: tuple[Any, ...],
|
|
limit: int,
|
|
lease_seconds: int,
|
|
) -> list[dict[str, Any]]:
|
|
allowed = {
|
|
"status='AWAITING_DECISION' AND channel='workspace_agent' AND callback_state='AWAITING_CALLBACK' AND callback_deadline <= ?",
|
|
"status='POLLING_RESPONSE' AND next_attempt_at <= ?",
|
|
}
|
|
if where_sql not in allowed:
|
|
raise ValueError("invalid_event_claim_query")
|
|
now = int(time.time())
|
|
conn = self._conn()
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT event_id FROM events
|
|
WHERE {where_sql}
|
|
AND (lease_until IS NULL OR lease_until < ?)
|
|
ORDER BY updated_at
|
|
LIMIT ?
|
|
""",
|
|
(*where_params, now, limit),
|
|
).fetchall()
|
|
ids = [str(row["event_id"]) for row in rows]
|
|
for event_id in ids:
|
|
conn.execute(
|
|
"""
|
|
UPDATE events SET lease_owner=?, lease_until=?
|
|
WHERE event_id=?
|
|
""",
|
|
(owner, now + lease_seconds, event_id),
|
|
)
|
|
conn.execute("COMMIT")
|
|
except Exception:
|
|
conn.execute("ROLLBACK")
|
|
raise
|
|
if not ids:
|
|
return []
|
|
placeholders = ",".join("?" for _ in ids)
|
|
return [
|
|
dict(row)
|
|
for row in conn.execute(
|
|
f"SELECT * FROM events WHERE event_id IN ({placeholders})",
|
|
ids,
|
|
).fetchall()
|
|
]
|
|
|
|
def apply_decision(
|
|
self,
|
|
decision: Decision,
|
|
*,
|
|
max_hops_per_gate: int,
|
|
max_same_blocker_retries: int,
|
|
) -> int:
|
|
now = int(time.time())
|
|
conn = self._conn()
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
event = conn.execute(
|
|
"SELECT project_key FROM events WHERE event_id=?",
|
|
(decision.event_id,),
|
|
).fetchone()
|
|
if event is None:
|
|
raise ConflictError("event_not_found")
|
|
if event["project_key"] != decision.project_key:
|
|
raise ConflictError("event_project_mismatch")
|
|
|
|
existing = conn.execute(
|
|
"SELECT decision_json, state_version_after FROM decisions WHERE event_id=?",
|
|
(decision.event_id,),
|
|
).fetchone()
|
|
if existing:
|
|
if json.loads(existing["decision_json"]) != decision.to_dict():
|
|
raise ConflictError("decision_idempotency_conflict")
|
|
conn.execute(
|
|
"UPDATE events SET callback_state='APPLIED', callback_deadline=NULL WHERE event_id=?",
|
|
(decision.event_id,),
|
|
)
|
|
conn.execute("COMMIT")
|
|
return int(existing["state_version_after"])
|
|
|
|
conn.execute(
|
|
"""
|
|
UPDATE events SET callback_state='CALLBACK_RECEIVED',
|
|
callback_received_at=?, callback_deadline=NULL
|
|
WHERE event_id=?
|
|
""",
|
|
(now, decision.event_id),
|
|
)
|
|
|
|
state = conn.execute(
|
|
"SELECT * FROM project_state WHERE project_key=?",
|
|
(decision.project_key,),
|
|
).fetchone()
|
|
if state is None:
|
|
raise ConflictError("project_state_not_found")
|
|
if state["contract_commit"] != decision.contract_commit:
|
|
raise ConflictError("contract_version_mismatch")
|
|
if state["state_version"] != decision.expected_state_version:
|
|
raise ConflictError("stale_decision_state_version")
|
|
state_json = json.loads(state["state_json"])
|
|
gate = str(
|
|
decision.state_updates.get(
|
|
"current_gate", state_json.get("current_gate", "unknown")
|
|
)
|
|
)
|
|
gate_hops = conn.execute(
|
|
"SELECT hops FROM gate_hops WHERE project_key=? AND gate_id=?",
|
|
(decision.project_key, gate),
|
|
).fetchone()
|
|
if gate_hops is not None and int(gate_hops["hops"]) >= max_hops_per_gate:
|
|
raise ConflictError("max_hops_exceeded")
|
|
|
|
state_json.update(decision.state_updates)
|
|
next_version = int(state["state_version"]) + 1
|
|
|
|
if decision.stop_reason:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT repetitions FROM blocker_history
|
|
WHERE project_key=? AND gate_id=? AND blocker=?
|
|
""",
|
|
(decision.project_key, gate, decision.stop_reason),
|
|
).fetchone()
|
|
repetitions = (int(row["repetitions"]) if row else 0) + 1
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO blocker_history(
|
|
project_key, gate_id, blocker, repetitions, updated_at
|
|
) VALUES (?, ?, ?, ?, ?)
|
|
ON CONFLICT(project_key, gate_id, blocker)
|
|
DO UPDATE SET repetitions=excluded.repetitions,
|
|
updated_at=excluded.updated_at
|
|
""",
|
|
(decision.project_key, gate, decision.stop_reason, repetitions, now),
|
|
)
|
|
if repetitions > max_same_blocker_retries:
|
|
raise ConflictError("same_blocker_retry_limit_exceeded")
|
|
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO gate_hops(project_key, gate_id, hops, updated_at)
|
|
VALUES (?, ?, 1, ?)
|
|
ON CONFLICT(project_key, gate_id)
|
|
DO UPDATE SET hops=gate_hops.hops + 1, updated_at=excluded.updated_at
|
|
""",
|
|
(decision.project_key, gate, now),
|
|
)
|
|
|
|
conn.execute(
|
|
"""
|
|
UPDATE project_state
|
|
SET state_version=?, state_json=?, hops=hops+1, updated_at=?
|
|
WHERE project_key=?
|
|
""",
|
|
(
|
|
next_version,
|
|
json.dumps(state_json, separators=(",", ":"), sort_keys=True),
|
|
now,
|
|
decision.project_key,
|
|
),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO decisions(
|
|
event_id, project_key, decision_json,
|
|
state_version_after, created_at
|
|
) VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
decision.event_id,
|
|
decision.project_key,
|
|
json.dumps(decision.to_dict(), separators=(",", ":"), sort_keys=True),
|
|
next_version,
|
|
now,
|
|
),
|
|
)
|
|
for unit in decision.next_work_units:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO outbox(
|
|
work_unit_id, event_id, assigned_agent, payload_json,
|
|
status, next_attempt_at, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, 'PENDING', ?, ?, ?)
|
|
""",
|
|
(
|
|
unit.id,
|
|
decision.event_id,
|
|
unit.primary_agent,
|
|
json.dumps(unit.to_dict(), separators=(",", ":"), sort_keys=True),
|
|
now,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
UPDATE events SET status='DECIDED', callback_state='APPLIED',
|
|
applied_at=?, callback_deadline=NULL, lease_owner=NULL,
|
|
lease_until=NULL, updated_at=?
|
|
WHERE event_id=?
|
|
""",
|
|
(now, now, decision.event_id),
|
|
)
|
|
conn.execute("COMMIT")
|
|
return next_version
|
|
except Exception:
|
|
conn.execute("ROLLBACK")
|
|
raise
|
|
|
|
def claim_outbox(self, *, owner: str, limit: int, lease_seconds: int) -> list[dict[str, Any]]:
|
|
rows = self._claim_table(
|
|
table="outbox",
|
|
key="work_unit_id",
|
|
owner=owner,
|
|
statuses=("PENDING", "RETRY"),
|
|
limit=limit,
|
|
lease_seconds=lease_seconds,
|
|
)
|
|
for row in rows:
|
|
context = self._conn().execute(
|
|
"""
|
|
SELECT e.project_key, d.state_version_after
|
|
FROM events e JOIN decisions d ON d.event_id=e.event_id
|
|
WHERE e.event_id=?
|
|
""",
|
|
(row["event_id"],),
|
|
).fetchone()
|
|
if context is not None:
|
|
row["project_key"] = str(context["project_key"])
|
|
row["state_version"] = int(context["state_version_after"])
|
|
return rows
|
|
|
|
def event_status(self, event_id: str) -> dict[str, Any] | None:
|
|
row = self._conn().execute(
|
|
"""
|
|
SELECT e.event_id, e.status, e.channel, e.remote_id,
|
|
e.trigger_http_status, e.trigger_error_category,
|
|
e.callback_state, e.callback_deadline, e.callback_attempts,
|
|
d.state_version_after
|
|
FROM events e LEFT JOIN decisions d ON d.event_id=e.event_id
|
|
WHERE e.event_id=?
|
|
""",
|
|
(event_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
raw = str(row["status"])
|
|
if raw == "DECIDED":
|
|
public_status = "DECIDED"
|
|
elif raw == "DEAD_LETTER":
|
|
public_status = "DEAD_LETTER"
|
|
elif raw == "TRIGGER_FAILED":
|
|
public_status = "TRIGGER_FAILED"
|
|
elif raw in {"AWAITING_DECISION", "POLLING_RESPONSE"}:
|
|
public_status = "AWAITING_DECISION"
|
|
else:
|
|
public_status = "RECEIVED"
|
|
channel = row["channel"]
|
|
if channel not in {"workspace_agent", "responses_api"}:
|
|
channel = None
|
|
version = row["state_version_after"]
|
|
return {
|
|
"event_id": str(row["event_id"]),
|
|
"status": public_status,
|
|
"trigger_channel": channel,
|
|
"trigger_http_status": (
|
|
None if row["trigger_http_status"] is None
|
|
else int(row["trigger_http_status"])
|
|
),
|
|
"trigger_error_category": row["trigger_error_category"],
|
|
"remote_id": row["remote_id"],
|
|
"callback_state": row["callback_state"],
|
|
"callback_deadline": row["callback_deadline"],
|
|
"callback_attempts": int(row["callback_attempts"]),
|
|
"decision_applied": raw == "DECIDED" and version is not None,
|
|
"state_version_after": None if version is None else int(version),
|
|
}
|
|
|
|
def mark_callback_state(self, event_id: str, state: str) -> None:
|
|
if state not in CALLBACK_STATES:
|
|
raise ValueError("invalid_callback_state")
|
|
now = int(time.time())
|
|
received = now if state == "CALLBACK_RECEIVED" else None
|
|
applied = now if state == "APPLIED" else None
|
|
self._conn().execute(
|
|
"""
|
|
UPDATE events SET callback_state=?,
|
|
callback_received_at=COALESCE(?, callback_received_at),
|
|
applied_at=COALESCE(?, applied_at),
|
|
callback_deadline=NULL, lease_owner=NULL, lease_until=NULL,
|
|
updated_at=? WHERE event_id=?
|
|
""",
|
|
(state, received, applied, now, event_id),
|
|
)
|
|
|
|
def pending_decision_plan(self, contract_commit: str, max_retries: int) -> dict[str, int]:
|
|
counts = {key: 0 for key in (
|
|
"eligible", "stale", "duplicate", "contract_mismatch",
|
|
"state_version_mismatch", "dead_letter_candidate",
|
|
)}
|
|
now = int(time.time())
|
|
rows = self._conn().execute(
|
|
"""
|
|
SELECT e.*, p.contract_commit AS current_contract,
|
|
p.state_version AS current_state_version,
|
|
d.event_id AS decision_exists
|
|
FROM events e
|
|
JOIN project_state p ON p.project_key=e.project_key
|
|
LEFT JOIN decisions d ON d.event_id=e.event_id
|
|
WHERE e.callback_state IN ('PENDING','SENT','AWAITING_CALLBACK','CALLBACK_RECEIVED')
|
|
"""
|
|
).fetchall()
|
|
for row in rows:
|
|
event = json.loads(row["event_json"])
|
|
if row["decision_exists"] is not None:
|
|
counts["duplicate"] += 1
|
|
elif event["contract_commit"] != contract_commit or row["current_contract"] != contract_commit:
|
|
counts["contract_mismatch"] += 1
|
|
elif int(event["state_version"]) != int(row["current_state_version"]):
|
|
counts["state_version_mismatch"] += 1
|
|
elif int(row["callback_attempts"]) >= max_retries:
|
|
counts["dead_letter_candidate"] += 1
|
|
elif row["callback_deadline"] is None or int(row["callback_deadline"]) <= now:
|
|
counts["eligible"] += 1
|
|
else:
|
|
counts["stale"] += 1
|
|
return counts
|
|
|
|
def recover_one_pending(self, *, owner: str, contract_commit: str,
|
|
lease_seconds: int, max_retries: int,
|
|
retry_after: int) -> str | None:
|
|
now = int(time.time())
|
|
conn = self._conn()
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT e.event_id, e.event_json, e.callback_attempts,
|
|
p.contract_commit AS current_contract,
|
|
p.state_version AS current_state_version,
|
|
d.event_id AS decision_exists
|
|
FROM events e JOIN project_state p ON p.project_key=e.project_key
|
|
LEFT JOIN decisions d ON d.event_id=e.event_id
|
|
WHERE e.callback_state IN ('PENDING','SENT','AWAITING_CALLBACK','CALLBACK_RECEIVED')
|
|
AND (e.lease_until IS NULL OR e.lease_until < ?)
|
|
AND (
|
|
e.callback_deadline IS NULL OR e.callback_deadline <= ?
|
|
OR d.event_id IS NOT NULL
|
|
OR p.contract_commit != ?
|
|
OR CAST(json_extract(e.event_json, '$.state_version') AS INTEGER)
|
|
!= p.state_version
|
|
OR e.callback_attempts >= ?
|
|
)
|
|
ORDER BY e.created_at LIMIT 1
|
|
""",
|
|
(now, now, contract_commit, max_retries),
|
|
).fetchone()
|
|
if row is None:
|
|
conn.execute("COMMIT")
|
|
return None
|
|
event = json.loads(row["event_json"])
|
|
state = "AWAITING_CALLBACK"
|
|
if row["decision_exists"] is not None:
|
|
state = "APPLIED"
|
|
elif event["contract_commit"] != contract_commit or row["current_contract"] != contract_commit:
|
|
state = "STALE_SUPERSEDED"
|
|
elif int(event["state_version"]) != int(row["current_state_version"]):
|
|
state = "STALE_SUPERSEDED"
|
|
elif int(row["callback_attempts"]) >= max_retries:
|
|
state = "DEAD_LETTER"
|
|
conn.execute(
|
|
"""
|
|
UPDATE events SET lease_owner=?, lease_until=?, callback_state=?,
|
|
callback_attempts=callback_attempts+1,
|
|
callback_deadline=?, updated_at=? WHERE event_id=?
|
|
""",
|
|
(owner, now + lease_seconds, state,
|
|
now + retry_after if state == "AWAITING_CALLBACK" else None,
|
|
now, row["event_id"]),
|
|
)
|
|
conn.execute("COMMIT")
|
|
self.fsync_state()
|
|
return str(row["event_id"])
|
|
except Exception:
|
|
conn.execute("ROLLBACK")
|
|
raise
|
|
|
|
def fsync_state(self) -> None:
|
|
self._conn().execute("PRAGMA wal_checkpoint(FULL)")
|
|
fd = os.open(self.path, os.O_RDONLY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
def mark_outbox(
|
|
self,
|
|
work_unit_id: str,
|
|
*,
|
|
status: str,
|
|
retry_after: int = 0,
|
|
increment_attempt: bool = False,
|
|
) -> None:
|
|
self._mark_table(
|
|
table="outbox",
|
|
key="work_unit_id",
|
|
value=work_unit_id,
|
|
status=status,
|
|
retry_after=retry_after,
|
|
increment_attempt=increment_attempt,
|
|
)
|
|
|
|
def enqueue_lifecycle(
|
|
self,
|
|
*,
|
|
sync_id: str,
|
|
project_key: str,
|
|
work_unit_id: str | None,
|
|
kind: str,
|
|
payload: dict[str, Any],
|
|
) -> None:
|
|
now = int(time.time())
|
|
encoded = json.dumps(payload, separators=(",", ":"), sort_keys=True)
|
|
self._conn().execute(
|
|
"""
|
|
INSERT OR IGNORE INTO lifecycle(
|
|
sync_id, project_key, work_unit_id, kind, payload_json,
|
|
status, next_attempt_at, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, 'PENDING', ?, ?, ?)
|
|
""",
|
|
(sync_id, project_key, work_unit_id, kind, encoded, now, now, now),
|
|
)
|
|
|
|
def claim_lifecycle(self, *, owner: str, limit: int, lease_seconds: int) -> list[dict[str, Any]]:
|
|
return self._claim_table(
|
|
table="lifecycle",
|
|
key="sync_id",
|
|
owner=owner,
|
|
statuses=("PENDING", "RETRY"),
|
|
limit=limit,
|
|
lease_seconds=lease_seconds,
|
|
)
|
|
|
|
def mark_lifecycle(
|
|
self,
|
|
sync_id: str,
|
|
*,
|
|
status: str,
|
|
retry_after: int = 0,
|
|
increment_attempt: bool = False,
|
|
) -> None:
|
|
self._mark_table(
|
|
table="lifecycle",
|
|
key="sync_id",
|
|
value=sync_id,
|
|
status=status,
|
|
retry_after=retry_after,
|
|
increment_attempt=increment_attempt,
|
|
)
|
|
|
|
def _claim_table(
|
|
self,
|
|
*,
|
|
table: str,
|
|
key: str,
|
|
owner: str,
|
|
statuses: tuple[str, ...],
|
|
limit: int,
|
|
lease_seconds: int,
|
|
) -> list[dict[str, Any]]:
|
|
if table not in {"outbox", "lifecycle"} or key not in {"work_unit_id", "sync_id"}:
|
|
raise ValueError("invalid_claim_table")
|
|
now = int(time.time())
|
|
placeholders = ",".join("?" for _ in statuses)
|
|
conn = self._conn()
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT {key} FROM {table}
|
|
WHERE status IN ({placeholders})
|
|
AND next_attempt_at <= ?
|
|
AND (lease_until IS NULL OR lease_until < ?)
|
|
ORDER BY created_at
|
|
LIMIT ?
|
|
""",
|
|
(*statuses, now, now, limit),
|
|
).fetchall()
|
|
values = [row[key] for row in rows]
|
|
for item in values:
|
|
conn.execute(
|
|
f"""
|
|
UPDATE {table} SET lease_owner=?, lease_until=?, updated_at=?
|
|
WHERE {key}=?
|
|
""",
|
|
(owner, now + lease_seconds, now, item),
|
|
)
|
|
conn.execute("COMMIT")
|
|
except Exception:
|
|
conn.execute("ROLLBACK")
|
|
raise
|
|
if not values:
|
|
return []
|
|
placeholders = ",".join("?" for _ in values)
|
|
return [
|
|
dict(row)
|
|
for row in conn.execute(
|
|
f"SELECT * FROM {table} WHERE {key} IN ({placeholders})",
|
|
values,
|
|
).fetchall()
|
|
]
|
|
|
|
def _mark_table(
|
|
self,
|
|
*,
|
|
table: str,
|
|
key: str,
|
|
value: str,
|
|
status: str,
|
|
retry_after: int,
|
|
increment_attempt: bool,
|
|
) -> None:
|
|
if table not in {"outbox", "lifecycle"} or key not in {"work_unit_id", "sync_id"}:
|
|
raise ValueError("invalid_mark_table")
|
|
now = int(time.time())
|
|
self._conn().execute(
|
|
f"""
|
|
UPDATE {table}
|
|
SET status=?, attempts=attempts + ?, next_attempt_at=?,
|
|
lease_owner=NULL, lease_until=NULL, updated_at=?
|
|
WHERE {key}=?
|
|
""",
|
|
(
|
|
status,
|
|
1 if increment_attempt else 0,
|
|
now + max(0, retry_after),
|
|
now,
|
|
value,
|
|
),
|
|
)
|
|
|
|
def project_state(self, project_key: str) -> dict[str, Any] | None:
|
|
row = self._conn().execute(
|
|
"SELECT * FROM project_state WHERE project_key=?",
|
|
(project_key,),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
result = dict(row)
|
|
result["state"] = json.loads(result.pop("state_json"))
|
|
return result
|
|
|
|
def all_project_states(self) -> list[dict[str, Any]]:
|
|
result = []
|
|
for row in self._conn().execute("SELECT * FROM project_state ORDER BY project_key").fetchall():
|
|
item = dict(row)
|
|
item["state"] = json.loads(item.pop("state_json"))
|
|
result.append(item)
|
|
return result
|
|
|
|
def summary_due(self, project_key: str, interval: int) -> bool:
|
|
now = int(time.time())
|
|
row = self._conn().execute(
|
|
"SELECT last_emitted_at FROM summaries WHERE project_key=?",
|
|
(project_key,),
|
|
).fetchone()
|
|
return row is None or now - int(row["last_emitted_at"]) >= interval
|
|
|
|
def mark_summary(self, project_key: str, digest: str) -> None:
|
|
now = int(time.time())
|
|
self._conn().execute(
|
|
"""
|
|
INSERT INTO summaries(project_key, last_emitted_at, last_digest)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(project_key)
|
|
DO UPDATE SET last_emitted_at=excluded.last_emitted_at,
|
|
last_digest=excluded.last_digest
|
|
""",
|
|
(project_key, now, digest),
|
|
)
|
|
|
|
|
|
def project_summary(self, project_key: str) -> dict[str, Any] | None:
|
|
state = self.project_state(project_key)
|
|
if state is None:
|
|
return None
|
|
conn = self._conn()
|
|
event_counts = {
|
|
row["status"]: int(row["count"])
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT status, COUNT(*) AS count FROM events
|
|
WHERE project_key=? GROUP BY status
|
|
""",
|
|
(project_key,),
|
|
).fetchall()
|
|
}
|
|
outbox_counts = {
|
|
row["status"]: int(row["count"])
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT o.status, COUNT(*) AS count
|
|
FROM outbox o JOIN events e ON e.event_id=o.event_id
|
|
WHERE e.project_key=? GROUP BY o.status
|
|
""",
|
|
(project_key,),
|
|
).fetchall()
|
|
}
|
|
return {
|
|
"project_key": project_key,
|
|
"contract_commit": state["contract_commit"],
|
|
"state_version": state["state_version"],
|
|
"state": state["state"],
|
|
"event_counts": event_counts,
|
|
"work_unit_counts": outbox_counts,
|
|
"updated_at": state["updated_at"],
|
|
}
|
|
|
|
|
|
def metadata_get(self, key: str) -> str | None:
|
|
row = self._conn().execute(
|
|
"SELECT value FROM metadata WHERE key=?",
|
|
(key,),
|
|
).fetchone()
|
|
return None if row is None else str(row["value"])
|
|
|
|
def metadata_set(self, key: str, value: str) -> None:
|
|
self._conn().execute(
|
|
"""
|
|
INSERT INTO metadata(key, value) VALUES (?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value=excluded.value
|
|
""",
|
|
(key, value),
|
|
)
|
|
|
|
def observe_refs(self, refs: dict[str, str]) -> tuple[bool, list[tuple[str, str | None, str]]]:
|
|
"""Persist a Git snapshot and return changes after the initial baseline."""
|
|
now = int(time.time())
|
|
conn = self._conn()
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
initialized = conn.execute(
|
|
"SELECT value FROM metadata WHERE key='git_poll_initialized'"
|
|
).fetchone()
|
|
previous = {
|
|
row["refname"]: row["object_id"]
|
|
for row in conn.execute("SELECT refname, object_id FROM observed_refs").fetchall()
|
|
}
|
|
changes: list[tuple[str, str | None, str]] = []
|
|
if initialized is not None:
|
|
for refname, oid in refs.items():
|
|
old = previous.get(refname)
|
|
if old != oid:
|
|
changes.append((refname, old, oid))
|
|
for refname, old in previous.items():
|
|
if refname not in refs:
|
|
changes.append((refname, old, "0" * 40))
|
|
conn.execute("DELETE FROM observed_refs")
|
|
for refname, oid in refs.items():
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO observed_refs(refname, object_id, observed_at)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(refname, oid, now),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO metadata(key, value)
|
|
VALUES ('git_poll_initialized', '1')
|
|
ON CONFLICT(key) DO UPDATE SET value='1'
|
|
"""
|
|
)
|
|
conn.execute("COMMIT")
|
|
return initialized is not None, changes
|
|
except Exception:
|
|
conn.execute("ROLLBACK")
|
|
raise
|
|
|
|
|
|
def close(self) -> None:
|
|
conn = getattr(self._local, "conn", None)
|
|
if conn is not None:
|
|
conn.close()
|
|
self._local.conn = None
|