165 lines
6.1 KiB
Python
165 lines
6.1 KiB
Python
"""Tenant-partitioned Hermes shared agent fabric primitives."""
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from contextlib import closing
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from typing import Any, TypeVar
|
|
|
|
from .tenant_context import (
|
|
CanonicalMessage, ConflictError, ScopeError, ValidationError,
|
|
isoformat, require_ident,
|
|
)
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
@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)
|
|
Path(self.database).parent.mkdir(parents=True, exist_ok=True)
|
|
self._initialize()
|
|
os.chmod(self.database, 0o600)
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
db = sqlite3.connect(self.database, timeout=30)
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
db.execute("PRAGMA synchronous=FULL")
|
|
db.execute("PRAGMA busy_timeout=30000")
|
|
return db
|
|
|
|
def _storage(self, operation: Callable[[], T]) -> T:
|
|
failure: ConflictError | None = None
|
|
try:
|
|
return operation()
|
|
except sqlite3.Error:
|
|
failure = ConflictError("fabric storage unavailable")
|
|
raise failure
|
|
|
|
def _initialize(self) -> None:
|
|
def initialize() -> None:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
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)"
|
|
)
|
|
self._storage(initialize)
|
|
|
|
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()
|
|
failure: ConflictError | None = None
|
|
try:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
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:
|
|
failure = ConflictError(
|
|
"fabric event already exists or violates queue uniqueness"
|
|
)
|
|
except sqlite3.Error:
|
|
failure = ConflictError("fabric storage unavailable")
|
|
if failure is not None:
|
|
raise failure
|
|
return job
|
|
|
|
def claim(
|
|
self, tenant_id: str, cell_id: str, profile_id: str,
|
|
) -> FabricJob | None:
|
|
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
|
|
cell_id = require_ident("cell_id", cell_id)
|
|
profile_id = require_ident("profile_id", profile_id)
|
|
|
|
def claim_one() -> FabricJob | None:
|
|
with closing(self._connect()) 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
|
|
updated = db.execute(
|
|
"UPDATE fabric_jobs SET status='running', updated_at=? "
|
|
"WHERE job_id=? AND status='queued'",
|
|
(isoformat(), row[0]),
|
|
)
|
|
if updated.rowcount != 1:
|
|
db.rollback()
|
|
return None
|
|
db.commit()
|
|
return FabricJob(
|
|
job_id=row[0], tenant_id=row[1], cell_id=row[2],
|
|
profile_id=row[3], session_id=row[4], event_id=row[5],
|
|
status="running",
|
|
)
|
|
|
|
return self._storage(claim_one)
|
|
|
|
def health(self) -> dict[str, Any]:
|
|
def query() -> int:
|
|
with closing(self._connect()) as connection, connection as db:
|
|
return int(db.execute("SELECT count(*) FROM fabric_jobs").fetchone()[0])
|
|
|
|
return {
|
|
"component": "hermes_fabric", "jobs": self._storage(query),
|
|
"checked_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|