feat(036): agent run tracker — durable HTTP client for backend run events
This commit is contained in:
186
agent/src/ss_tools/agent/_run_tracker.py
Normal file
186
agent/src/ss_tools/agent/_run_tracker.py
Normal file
@@ -0,0 +1,186 @@
|
||||
# agent/src/ss_tools/agent/_run_tracker.py
|
||||
# #region AgentChat.RunTracker [C:4] [TYPE Module] [SEMANTICS agent-run,tracker,durable,emit]
|
||||
# @ingroup AgentChat
|
||||
# @BRIEF Durable run tracker: create backend AgentRun, append events, register drafts.
|
||||
# @LAYER Service
|
||||
# @RELATION DEPENDS_ON -> [AgentChat.Context]
|
||||
# @INVARIANT Backend persistence precedes Gradio yield — dropped streams remain recoverable.
|
||||
# @RATIONALE A separate tracker client keeps run durability independent of the chat stream lifecycle.
|
||||
# @REJECTED Embedding run tracking into app.py — rejected because run state must survive Gradio restarts.
|
||||
import hashlib
|
||||
import json
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from ss_tools.shared.logger import logger
|
||||
|
||||
|
||||
def _hash_payload(data: dict[str, Any] | None) -> str:
|
||||
if data is None:
|
||||
data = {}
|
||||
return hashlib.sha256(
|
||||
json.dumps(data, sort_keys=True, ensure_ascii=False).encode("utf-8")
|
||||
).hexdigest()
|
||||
|
||||
|
||||
class RunTracker:
|
||||
"""Client for the backend AgentRuns API.
|
||||
|
||||
Talks to the FastAPI backend over HTTP using the service JWT.
|
||||
Persists run lifecycle, stage progress, drafts, and approval gates.
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: str, service_jwt: str):
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._jwt = service_jwt
|
||||
self._client: httpx.AsyncClient | None = None
|
||||
self._run_id: str | None = None
|
||||
self._sequence: int = 0
|
||||
|
||||
async def _ensure_client(self) -> httpx.AsyncClient:
|
||||
if self._client is None:
|
||||
self._client = httpx.AsyncClient(
|
||||
headers={"Authorization": f"Bearer {self._jwt}"},
|
||||
timeout=httpx.Timeout(15.0),
|
||||
)
|
||||
return self._client
|
||||
|
||||
# ── Create run ─────────────────────────────────────────────
|
||||
|
||||
async def create(
|
||||
self,
|
||||
context: dict[str, Any],
|
||||
conversation_id: str | None = None,
|
||||
) -> str:
|
||||
"""Create a durable AgentRun and return the run_id.
|
||||
|
||||
Call this before any scenario tool action begins.
|
||||
"""
|
||||
client = await self._ensure_client()
|
||||
idempotency_key = str(uuid.uuid4())
|
||||
try:
|
||||
resp = await client.post(
|
||||
f"{self._base_url}/api/agent/runs",
|
||||
json={
|
||||
"context": context,
|
||||
"conversation_id": conversation_id,
|
||||
"idempotency_key": idempotency_key,
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
logger.explore(
|
||||
"Failed to create agent run",
|
||||
{"status": exc.response.status_code, "body": exc.response.text[:500]},
|
||||
)
|
||||
raise
|
||||
|
||||
data = resp.json()
|
||||
self._run_id = data["id"]
|
||||
self._sequence = data.get("last_sequence", 0)
|
||||
logger.reason(
|
||||
"Agent run created",
|
||||
{"run_id": self._run_id, "sequence": self._sequence},
|
||||
)
|
||||
return self._run_id
|
||||
|
||||
# ── Append event ───────────────────────────────────────────
|
||||
|
||||
async def append_event(
|
||||
self,
|
||||
event_type: str,
|
||||
stage: str | None = None,
|
||||
status: str | None = None,
|
||||
payload: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Append a typed event to the backend run.
|
||||
|
||||
Backend persistence MUST succeed before Gradio yield.
|
||||
"""
|
||||
if self._run_id is None:
|
||||
raise RuntimeError("Must call create() before append_event()")
|
||||
self._sequence += 1
|
||||
client = await self._ensure_client()
|
||||
try:
|
||||
resp = await client.post(
|
||||
f"{self._base_url}/api/agent/runs/{self._run_id}/events",
|
||||
json={
|
||||
"event_type": event_type,
|
||||
"stage": stage,
|
||||
"status": status,
|
||||
"sequence": self._sequence,
|
||||
"payload": payload,
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
logger.explore(
|
||||
"Failed to append agent run event — run remains recoverable from last snapshot",
|
||||
{"run_id": self._run_id, "sequence": self._sequence, "status": exc.response.status_code},
|
||||
)
|
||||
raise
|
||||
|
||||
logger.reason(
|
||||
"Agent run event appended",
|
||||
{"run_id": self._run_id, "event_type": event_type, "sequence": self._sequence},
|
||||
)
|
||||
return resp.json()
|
||||
|
||||
async def emit_progress(
|
||||
self,
|
||||
stage: str,
|
||||
status: str = "completed",
|
||||
) -> None:
|
||||
"""Emit a progress event for a scenario stage."""
|
||||
await self.append_event(
|
||||
event_type="progress",
|
||||
stage=stage,
|
||||
status=status,
|
||||
)
|
||||
|
||||
async def emit_draft(
|
||||
self,
|
||||
kind: str,
|
||||
name: str,
|
||||
intended_path: str,
|
||||
sha256: str,
|
||||
validation_status: str = "pending",
|
||||
) -> None:
|
||||
"""Register a draft artifact — emitted as progress event so frontend can update."""
|
||||
await self.append_event(
|
||||
event_type="drafts_updated",
|
||||
payload={
|
||||
"kind": kind,
|
||||
"name": name,
|
||||
"intended_path": intended_path,
|
||||
"sha256": sha256,
|
||||
"validation_status": validation_status,
|
||||
},
|
||||
)
|
||||
|
||||
async def emit_terminal(
|
||||
self,
|
||||
status: str,
|
||||
error_code: str | None = None,
|
||||
error_detail: str | None = None,
|
||||
) -> None:
|
||||
"""Mark the run as COMPLETED, FAILED, or CANCELLED."""
|
||||
await self.append_event(
|
||||
event_type="terminal",
|
||||
status=status,
|
||||
payload={"error_code": error_code, "error_detail": error_detail},
|
||||
)
|
||||
|
||||
@property
|
||||
def run_id(self) -> str | None:
|
||||
return self._run_id
|
||||
|
||||
# ── Cleanup ────────────────────────────────────────────────
|
||||
|
||||
async def close(self) -> None:
|
||||
if self._client:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
# #endregion AgentChat.RunTracker
|
||||
Reference in New Issue
Block a user