144 lines
5.0 KiB
Python
144 lines
5.0 KiB
Python
"""Tenant-partitioned Hermes shared agent fabric primitives."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from contextlib import closing
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from typing import Any
|
|
|
|
from .tenant_context import (
|
|
CanonicalMessage,
|
|
ConflictError,
|
|
ScopeError,
|
|
ValidationError,
|
|
isoformat,
|
|
require_ident,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FabricJob:
|
|
job_id: str
|
|
tenant_id: str
|
|
cell_id: str
|
|
profile_id: str
|
|
session_id: str
|
|
event_id: str
|
|
status: str = "queued"
|
|
|
|
def __post_init__(self) -> None:
|
|
for name in (
|
|
"job_id", "cell_id", "profile_id", "session_id", "event_id"
|
|
):
|
|
require_ident(name, getattr(self, name))
|
|
require_ident("tenant_id", self.tenant_id, tenant=True)
|
|
if self.status not in {"queued", "running", "completed", "failed", "cancelled"}:
|
|
raise ValidationError("invalid job status")
|
|
|
|
|
|
class HermesFabric:
|
|
"""Shared scheduler storage with mandatory tenant partition keys."""
|
|
|
|
def __init__(self, database: str | Path) -> None:
|
|
self.database = str(database)
|
|
with closing(sqlite3.connect(self.database)) as connection, connection as db:
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
db.execute("PRAGMA synchronous=FULL")
|
|
db.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS fabric_jobs (
|
|
job_id TEXT PRIMARY KEY,
|
|
tenant_id TEXT NOT NULL,
|
|
cell_id TEXT NOT NULL,
|
|
profile_id TEXT NOT NULL,
|
|
session_id TEXT NOT NULL,
|
|
event_id TEXT NOT NULL UNIQUE,
|
|
status TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
db.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_fabric_tenant_queue "
|
|
"ON fabric_jobs(tenant_id, cell_id, status)"
|
|
)
|
|
|
|
def enqueue(
|
|
self,
|
|
*,
|
|
message: CanonicalMessage,
|
|
profile_id: str,
|
|
) -> FabricJob:
|
|
profile_id = require_ident("profile_id", profile_id)
|
|
if not message.command_capable:
|
|
raise ScopeError("non-command events cannot create execution jobs")
|
|
if not message.is_authenticated_human_command:
|
|
raise ScopeError("only authenticated human commands can create jobs")
|
|
job = FabricJob(
|
|
job_id=f"job-{message.event_id}",
|
|
tenant_id=message.tenant_id,
|
|
cell_id=message.cell_id,
|
|
profile_id=profile_id,
|
|
session_id=message.session_id,
|
|
event_id=message.event_id,
|
|
)
|
|
now = isoformat()
|
|
conflict: ConflictError | None = None
|
|
try:
|
|
with closing(sqlite3.connect(self.database)) as connection, connection as db:
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
db.execute(
|
|
"INSERT INTO fabric_jobs VALUES(?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
job.job_id,
|
|
job.tenant_id,
|
|
job.cell_id,
|
|
job.profile_id,
|
|
job.session_id,
|
|
job.event_id,
|
|
job.status,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
except sqlite3.IntegrityError:
|
|
conflict = ConflictError(
|
|
"fabric event already exists or violates queue uniqueness"
|
|
)
|
|
if conflict is not None:
|
|
raise conflict
|
|
return job
|
|
|
|
def claim(self, tenant_id: str, cell_id: str, profile_id: str) -> FabricJob | None:
|
|
require_ident("tenant_id", tenant_id, tenant=True)
|
|
require_ident("cell_id", cell_id)
|
|
require_ident("profile_id", profile_id)
|
|
with closing(sqlite3.connect(self.database)) as connection, connection as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute(
|
|
"SELECT * FROM fabric_jobs WHERE tenant_id=? AND cell_id=? "
|
|
"AND profile_id=? AND status='queued' ORDER BY created_at LIMIT 1",
|
|
(tenant_id, cell_id, profile_id),
|
|
).fetchone()
|
|
if not row:
|
|
db.commit()
|
|
return None
|
|
db.execute(
|
|
"UPDATE fabric_jobs SET status='running', updated_at=? WHERE job_id=?",
|
|
(isoformat(), row[0]),
|
|
)
|
|
db.commit()
|
|
return FabricJob(*row[:7])
|
|
|
|
def health(self) -> dict[str, Any]:
|
|
with closing(sqlite3.connect(self.database)) as connection, connection as db:
|
|
count = db.execute("SELECT count(*) FROM fabric_jobs").fetchone()[0]
|
|
return {
|
|
"component": "hermes_fabric",
|
|
"jobs": count,
|
|
"checked_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|