feat(hermes): complete HUX backend families

This commit is contained in:
jenkins 2026-08-24 02:24:12 -03:00
parent a1070449a7
commit 53d7c2c586
28 changed files with 2963 additions and 34 deletions

View File

@ -28,10 +28,12 @@ SCHEMA_FILES = (
"artifact.schema.json",
"permission.schema.json",
"mode.schema.json",
"multimodal.schema.json",
"citation.schema.json",
"suggestion.schema.json",
"privacy.schema.json",
"release.schema.json",
"release-ledger.schema.json",
)
SUPPORTED_KEYWORDS = frozenset(
{
@ -75,7 +77,7 @@ def load_flags(directory: Path = CONTRACT_DIR) -> dict[str, Any]:
def _walk(node: Any, path: str, problems: list[str]) -> None:
if isinstance(node, dict):
for key, value in node.items():
if path.endswith("/properties") or path.endswith("/$defs"):
if path.endswith(("/properties", "/$defs")):
_walk(value, f"{path}/{key}", problems)
continue
if key not in SUPPORTED_KEYWORDS:

View File

@ -19,20 +19,32 @@ from hux.rules import flag_enabled, flag_registry
CONTRACT_VERSION = "1.1.0"
CARD_ROUTES: dict[str, list[str]] = {
"HUX-11": ["/hux/v1/capabilities", "/hux/v1/manifest"],
"HUX-11": ["/hux/v1/capabilities", "/hux/v1/manifest", "/hux/v1/context/bootstrap"],
"HUX-01": ["/hux/v1/conversations/{id}/events", "/hux/v1/conversations/{id}/events/stream"],
"HUX-02": ["/hux/v1/memory", "/hux/v1/memory/{id}", "/hux/v1/memory/{id}/{action}", "/hux/v1/memory/export"],
"HUX-03": ["/hux/v1/projects", "/hux/v1/projects/{id}", "/hux/v1/conversations", "/hux/v1/conversations/{id}", "/hux/v1/conversations/{id}/branch", "/hux/v1/conversations/{id}/lineage", "/hux/v1/search"],
"HUX-04": ["/hux/v1/artifacts", "/hux/v1/artifacts/{id}", "/hux/v1/artifacts/{id}/versions", "/hux/v1/artifacts/{id}/versions/{n}", "/hux/v1/artifacts/{id}/versions/{n}/diff", "/hux/v1/artifacts/{id}/promote"],
"HUX-05": ["/hux/v1/policy", "/hux/v1/approvals", "/hux/v1/approvals/{id}", "/hux/v1/runs/{id}/stop", "/hux/v1/runs/{id}/budget", "/hux/v1/runs/{id}/gate"],
"HUX-06": [],
"HUX-07": [],
"HUX-06": ["/hux/v1/modes", "/hux/v1/projects/{project_id}/conversations/{id}/mode"],
"HUX-07": [
"/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items",
"/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items/{item_id}",
"/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items/{item_id}/transcript-corrections",
"/hux/v1/projects/{project_id}/conversations/{id}/capture-intents",
],
"HUX-08": ["/hux/v1/sources", "/hux/v1/sources/{id}", "/hux/v1/passages", "/hux/v1/messages/{id}/citations", "/hux/v1/notebooks", "/hux/v1/notebooks/{id}"],
"HUX-09": [],
"HUX-09": [
"/hux/v1/projects/{project_id}/conversations/{id}/suggestions/evaluate",
"/hux/v1/projects/{project_id}/conversations/{id}/suggestions/{suggestion_id}/decisions",
"/hux/v1/projects/{project_id}/conversations/{id}/suggestions/states",
],
"HUX-10": ["/hux/v1/privacy/policy", "/hux/v1/privacy/notices", "/hux/v1/conversations/{id}/forget", "/hux/v1/privacy/audit", "/hux/v1/conversations/{id}/privacy"],
"HUX-12": [],
"HUX-12": [
"/hux/v1/projects/{project_id}/conversations/{id}/releases",
"/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}",
"/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}/transitions",
],
}
# Cards whose routes are not shipped yet declare [] above until they land (F12).
# The only (method, template) pairs a ``trust: worker`` caller may reach (SO-08).
# Everything else is 403 before the flag check. These are the agent-hook
@ -42,6 +54,7 @@ CARD_ROUTES: dict[str, list[str]] = {
# private mode (SO-28) before proposing a memory.
WORKER_ROUTES: frozenset[tuple[str, str]] = frozenset({
("GET", "/hux/v1/capabilities"), ("GET", "/hux/v1/manifest"),
("POST", "/hux/v1/context/bootstrap"),
("POST", "/hux/v1/approvals"),
("POST", "/hux/v1/runs/{id}/gate"), ("POST", "/hux/v1/runs/{id}/budget"), ("POST", "/hux/v1/runs/{id}/stop"),
("GET", "/hux/v1/runs/{id}/budget"),
@ -50,6 +63,13 @@ WORKER_ROUTES: frozenset[tuple[str, str]] = frozenset({
("GET", "/hux/v1/memory"), ("POST", "/hux/v1/memory"),
("POST", "/hux/v1/sources"), ("POST", "/hux/v1/passages"), ("POST", "/hux/v1/messages/{id}/citations"),
("POST", "/hux/v1/artifacts"), ("POST", "/hux/v1/artifacts/{id}/versions"),
("GET", "/hux/v1/modes"),
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/mode"),
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items"),
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}"),
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}/transitions"),
})

View File

@ -1,9 +1,169 @@
"""HUX-11 routes: capability negotiation and the data manifest."""
"""HUX-11 routes: capabilities, manifest, and trusted context bootstrap."""
from __future__ import annotations
import hashlib
import hmac
import os
import re
import stat
from pathlib import Path
from typing import Any
from hux import contracts, organization
from hux.errors import Conflict, Forbidden, Invalid
from hux.flags import CONTRACT_VERSION
from hux.http import Request, Response, Router
from hux.store import check_id, now_iso
CONTEXT_FAMILY = "context_bindings"
CONTEXT_SCHEMA = "hux.context_bootstrap.v1"
CONTEXT_KEY_BYTES = 32
CONTEXT_MESSAGE = b"hux.context.id.v1"
RAW_RE = re.compile(r"^[A-Za-z0-9._:/@+\-]{1,240}$")
CONTEXT_ID_RE = re.compile(r"^(ses|conv|prj)_[0-9a-f]{32}$")
SCHEMAS = contracts.load_all()
def _context_key(environ: dict[str, str]) -> bytes:
"""Read the exact 0600 regular key file; inline keys and links fail closed."""
raw_path = environ.get("HUX_CONTEXT_KEY_FILE", "")
if not raw_path:
raise Forbidden("context identity key is unavailable")
path = Path(raw_path)
try:
metadata = path.lstat()
except OSError as error:
raise Forbidden("context identity key is unavailable") from error
if not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_uid != os.geteuid():
raise Forbidden("context identity key is unsafe")
try:
value = path.read_bytes()
except OSError as error:
raise Forbidden("context identity key is unavailable") from error
if len(value) != CONTEXT_KEY_BYTES:
raise Forbidden("context identity key is invalid")
return value
def _source(body: dict[str, Any], name: str) -> str:
value = body.get(name)
if not isinstance(value, str) or not RAW_RE.fullmatch(value):
raise Invalid(f"{name} is malformed")
return value
def derive_context_id(key: bytes, purpose: str, identity: Any, raw: str) -> str:
"""Derive one frozen HMAC-bound context id."""
prefix = {"session": "ses", "conversation": "conv", "project": "prj"}[purpose]
message = b"\0".join(
(CONTEXT_MESSAGE, purpose.encode(), identity.tenant_slot.encode(), identity.subject.encode(), raw.encode())
)
return f"{prefix}_{hmac.new(key, message, hashlib.sha256).hexdigest()[:32]}"
def _exact_body(request: Request) -> tuple[dict[str, Any], str, str]:
if not isinstance(request.body, dict):
raise Invalid("body must be a JSON object")
expected = {"raw_session_id", "project_source", "session_id", "conversation_id", "project_id"}
if set(request.body) != expected:
raise Invalid("context bootstrap fields are not exact")
raw_session = _source(request.body, "raw_session_id")
project_source = _source(request.body, "project_source")
for name, prefix in (("session_id", "ses"), ("conversation_id", "conv"), ("project_id", "prj")):
value = request.body.get(name)
if not isinstance(value, str) or not CONTEXT_ID_RE.fullmatch(value) or not value.startswith(f"{prefix}_"):
raise Invalid(f"{name} is malformed")
check_id(value)
return request.body, raw_session, project_source
def _verify_record(record: dict[str, Any], schema: str, owner: str, project_id: str | None = None) -> None:
if record.get("schema") != schema or record.get("owner") != owner:
raise Conflict("context identity does not match existing record")
if project_id is not None and record.get("project_id") != project_id:
raise Conflict("context linkage does not match existing record")
if contracts.validate_record(record, SCHEMAS):
raise Conflict("existing context record is invalid")
def _response(request: Request, project: dict[str, Any], conversation: dict[str, Any], created: dict[str, bool]) -> Response:
body = {
"schema": CONTEXT_SCHEMA,
"contract_version": CONTRACT_VERSION,
"identity": request.identity.record(),
"session_id": request.body["session_id"],
"conversation_id": conversation["id"],
"project_id": project["id"],
"created": created,
"revisions": {"project": project["revision"], "conversation": conversation["revision"]},
}
return Response(201 if any(created.values()) else 200, body)
def bootstrap_context(request: Request) -> Response:
"""Create deterministic server-owned project and conversation records for trusted runtimes."""
if request.identity.trust not in {"relay", "worker"}:
raise Forbidden("context bootstrap requires relay or worker trust")
body, raw_session, project_source = _exact_body(request)
idem = request.idempotency_key()
if not idem:
raise Invalid("Idempotency-Key is required")
key = _context_key(request.flags._environ)
expected = {
"session_id": derive_context_id(key, "session", request.identity, raw_session),
"conversation_id": derive_context_id(key, "conversation", request.identity, raw_session),
"project_id": derive_context_id(key, "project", request.identity, project_source),
}
if any(not hmac.compare_digest(body[name], value) for name, value in expected.items()):
raise Invalid("context identifiers do not match authenticated inputs")
with request.store.lock(CONTEXT_FAMILY):
binding = request.store.get(CONTEXT_FAMILY, body["session_id"]) if request.store.exists(CONTEXT_FAMILY, body["session_id"]) else None
if binding:
valid_binding = (
binding.get("schema") == "hux.context_binding.v1"
and binding.get("id") == body["session_id"]
and binding.get("owner") == request.identity.subject
and all(binding.get(name) == body[name] for name in expected)
)
if not valid_binding:
raise Conflict("context binding does not match existing linkage")
for row in request.store.read(CONTEXT_FAMILY, "idempotency"):
if row.get("key") == idem and any(row.get(name) != body[name] for name in expected):
raise Conflict("Idempotency-Key was used for another context")
project = request.store.get(organization.PROJECTS, body["project_id"]) if request.store.exists(organization.PROJECTS, body["project_id"]) else None
conversation = request.store.get(organization.CONVERSATIONS, body["conversation_id"]) if request.store.exists(organization.CONVERSATIONS, body["conversation_id"]) else None
if project:
_verify_record(project, "hux.project.v1", request.identity.subject)
if conversation:
_verify_record(conversation, "hux.conversation.v1", request.identity.subject, body["project_id"])
if project is None and request.store.count(organization.PROJECTS) >= organization.MAX_PROJECTS:
raise Conflict("projects cap reached")
if conversation is None and request.store.count(organization.CONVERSATIONS) >= organization.MAX_CONVERSATIONS:
raise Conflict("conversations cap reached")
stamp = now_iso()
created = {"project": project is None, "conversation": conversation is None}
project = project or request.store.put(organization.PROJECTS, organization.checked({
"schema": "hux.project.v1", "id": body["project_id"], "owner": request.identity.subject,
"name": "Hermes", "tags": [], "pinned": False, "archived": False, "default_mode": "fast",
"created_at": stamp, "updated_at": stamp, "revision": 1,
}))
conversation = conversation or request.store.put(organization.CONVERSATIONS, organization.checked({
"schema": "hux.conversation.v1", "id": body["conversation_id"], "owner": request.identity.subject,
"project_id": body["project_id"], "title": "Hermes conversation", "tags": [], "pinned": False,
"archived": False, "mode": "fast", "artifact_ids": [], "created_at": stamp, "updated_at": stamp,
"revision": 1,
}))
if binding is None:
request.store.put(CONTEXT_FAMILY, {
"schema": "hux.context_binding.v1", "id": body["session_id"], "owner": request.identity.subject,
"session_id": body["session_id"], "conversation_id": body["conversation_id"],
"project_id": body["project_id"], "created_at": stamp,
})
if not any(row.get("key") == idem for row in request.store.read(CONTEXT_FAMILY, "idempotency")):
request.store.append(CONTEXT_FAMILY, "idempotency", {"key": idem, **expected})
request.audit("foundation.bootstrap", body["session_id"], reason="created" if any(created.values()) else "replayed")
return _response(request, project, conversation, created)
def capabilities(request: Request) -> Response:
@ -22,3 +182,4 @@ def register(router: Router) -> None:
"""Attach HUX-11 routes."""
router.add("GET", "/hux/v1/capabilities", "HUX-11", "foundation.capabilities", capabilities)
router.add("GET", "/hux/v1/manifest", "HUX-11", "foundation.manifest", manifest)
router.add("POST", "/hux/v1/context/bootstrap", "HUX-11", "foundation.bootstrap", bootstrap_context, 4096)

View File

@ -14,11 +14,11 @@ import os
import re
import threading
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Callable
from collections.abc import Mapping
from typing import Any
from urllib.parse import parse_qs, urlsplit
from hux import audit
@ -252,7 +252,7 @@ def make_handler(router: Router) -> type[BaseHTTPRequestHandler]:
class HuxHandler(BaseHTTPRequestHandler):
server_version = "hux-foundation/1.0"
def log_message(self, fmt: str, *args: Any) -> None: # noqa: D102 - quiet by design; audit ledger is the log
def log_message(self, fmt: str, *args: Any) -> None:
return
def _run(self) -> None:
@ -275,7 +275,11 @@ def make_handler(router: Router) -> type[BaseHTTPRequestHandler]:
if len(body) != length:
raise Invalid("request body is incomplete or timed out")
if self.path == "/healthz":
self._send(Response(200, {"status": "ok", "contract_version": CONTRACT_VERSION}))
body = {"status": "ok", "contract_version": CONTRACT_VERSION}
scheduler = getattr(router, "retention_scheduler", None)
if scheduler is not None:
body["retention"] = scheduler.health()
self._send(Response(200, body))
return
self._send(router.dispatch(self.command, self.path, dict(self.headers.items()), body))
except HuxError as error:
@ -316,8 +320,9 @@ def make_handler(router: Router) -> type[BaseHTTPRequestHandler]:
class BoundedHTTPServer(ThreadingHTTPServer):
"""Threading server that bounds header and body socket reads from accept onward."""
def __init__(self, address: tuple[str, int], handler: type[BaseHTTPRequestHandler], timeout: float) -> None:
def __init__(self, address: tuple[str, int], handler: type[BaseHTTPRequestHandler], timeout: float, scheduler: Any = None) -> None:
self.request_timeout = timeout
self.scheduler = scheduler
super().__init__(address, handler)
def get_request(self) -> tuple[Any, Any]:
@ -326,9 +331,20 @@ class BoundedHTTPServer(ThreadingHTTPServer):
request.settimeout(self.request_timeout)
return request, address
def server_close(self) -> None:
"""Stop background work before closing the listening socket."""
if self.scheduler is not None:
self.scheduler.stop()
super().server_close()
def serve(router: Router, host: str = "127.0.0.1", port: int = 8790) -> ThreadingHTTPServer:
"""Create (but do not start) the server; callers call serve_forever()."""
server = BoundedHTTPServer((host, port), make_handler(router), router.request_timeout)
from hux.retention_scheduler import RetentionScheduler
scheduler = RetentionScheduler(router)
router.retention_scheduler = scheduler
server = BoundedHTTPServer((host, port), make_handler(router), router.request_timeout, scheduler)
server.daemon_threads = True
scheduler.start()
return server

View File

@ -13,8 +13,8 @@ import os
import re
import stat
import tempfile
from dataclasses import dataclass
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from hux.errors import Unauthorized
@ -154,8 +154,6 @@ def _enforce_subject_binding(environ: Mapping[str, str], trust: str, subject: st
bound = _read_subject_binding(path)
if bound is None or not hmac.compare_digest(bound, subject):
raise Unauthorized("subject does not match trusted binding")
def resolve(headers: Mapping[str, str], environ: Mapping[str, str] | None = None) -> Identity:
"""Build an Identity from request headers or raise Unauthorized.

View File

@ -0,0 +1,169 @@
"""HUX-06 provider-neutral friendly modes and scoped route selection.
Automatic modes express intent to Switchyard and never name a provider. An
exact route can only be selected through the explicit advanced path and must
exist in the operator-provided route catalog. Private mode is local-only.
"""
from __future__ import annotations
import hashlib
import json
import os
import re
from typing import Any
from hux import contracts, rules
from hux.errors import Conflict, Invalid, NotFound
from hux.http import Request, Response, Router, page
from hux.store import now_iso
CARD = "HUX-06"
FAMILY = "mode_selections"
MODE_NAMES = ("fast", "thoughtful", "research", "create", "private")
MANUAL_ROUTE = re.compile(r"^atlas/manual/(codex|claude|local)/[a-z0-9][a-z0-9/-]{0,100}$")
MAX_CATALOG_ROUTES = 256
SCHEMAS = contracts.load_all()
def _body(request: Request) -> dict[str, Any]:
if not isinstance(request.body, dict):
raise Invalid("body must be a JSON object")
extra = set(request.body) - {"project_id", "mode", "advanced", "override_route_id"}
if extra:
raise Invalid("unexpected mode fields", sorted(extra))
return request.body
def _scope(request: Request, project_id: Any) -> tuple[dict[str, Any], dict[str, Any]]:
"""Resolve a conversation and its mandatory owning project."""
from hux import organization
try:
conversation = request.store.get(organization.CONVERSATIONS, request.params["id"])
except (Invalid, NotFound) as error:
raise NotFound("conversation not found") from error
path_project = request.params.get("project_id")
actual = conversation.get("project_id")
if not isinstance(project_id, str) or project_id != path_project or not actual or project_id != actual:
raise NotFound("project or conversation not found")
try:
project = request.store.get(organization.PROJECTS, project_id)
except (Invalid, NotFound) as error:
raise NotFound("project or conversation not found") from error
return project, conversation
def _record_id(conversation_id: str) -> str:
digest = hashlib.sha256(conversation_id.encode()).hexdigest()[:24]
return f"mode_{digest}"
def _catalog() -> set[str]:
raw = os.environ.get("HUX_SWITCHYARD_ROUTE_CATALOG", "")
items = [item.strip() for item in raw.split(",") if item.strip()]
if len(items) > MAX_CATALOG_ROUTES:
return set()
return {item for item in items if MANUAL_ROUTE.fullmatch(item)}
def _contract(mode: str, advanced: bool, override: Any) -> dict[str, Any]:
if not isinstance(mode, str) or mode not in MODE_NAMES:
raise Invalid("unknown friendly mode")
if override is not None:
if not advanced or not isinstance(override, str) or not MANUAL_ROUTE.fullmatch(override):
raise Invalid("an exact route requires a valid advanced manual route")
if override not in _catalog():
raise Invalid("advanced route is not in the Switchyard catalog")
if mode == "private" and not override.startswith("atlas/manual/local/"):
raise Invalid("private mode is local-only")
elif advanced:
raise Invalid("advanced selection requires override_route_id")
try:
result = rules.mode_contract(mode, override)
except ValueError as error:
raise Invalid(str(error)) from error
problems = contracts.validate("mode.schema.json", result, SCHEMAS)
if problems:
raise Invalid("mode failed contract validation", problems)
if mode != "private" and result["switchyard"]["route_id"].startswith("atlas/manual/"):
raise Invalid("automatic modes may not pin a provider")
return result
def _fingerprint(body: dict[str, Any]) -> str:
allowed = {key: body.get(key) for key in ("project_id", "mode", "advanced", "override_route_id")}
return hashlib.sha256(json.dumps(allowed, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
def _replay(request: Request, key: str, fingerprint: str) -> dict[str, Any] | None:
for row in request.store.read(FAMILY, "idempotency"):
if row.get("key") != key:
continue
if row.get("fingerprint") != fingerprint:
raise Conflict("Idempotency-Key was already used for a different selection")
return request.store.get(FAMILY, row["id"])
return None
def list_modes(request: Request) -> Response:
"""Return all provider-neutral mode contracts."""
items = [_contract(name, False, None) for name in MODE_NAMES]
request.audit("modes.list", "modes")
return page(items)
def get_selection(request: Request) -> Response:
"""Return the selected mode for one bound conversation."""
project_id = request.params["project_id"]
_, conversation = _scope(request, project_id)
record = request.store.get(FAMILY, _record_id(conversation["id"]))
request.audit("modes.read", record["id"])
return Response(200, record, {"ETag": str(record["revision"]), "Cache-Control": "no-store"})
def put_selection(request: Request) -> Response:
"""Select a mode using If-Match and Idempotency-Key."""
body = _body(request)
_, conversation = _scope(request, body.get("project_id"))
key = request.idempotency_key()
if not key:
raise Invalid("Idempotency-Key is required")
expected = request.if_match()
if expected is None:
raise Invalid("If-Match is required; use 0 for the first selection")
fingerprint = _fingerprint(body)
mode = _contract(body.get("mode"), body.get("advanced") is True, body.get("override_route_id"))
record_id = _record_id(conversation["id"])
with request.store.lock(FAMILY):
replayed = _replay(request, key, fingerprint)
if replayed is not None:
request.audit("modes.select", replayed["id"], reason="idempotent_replay")
return Response(200, replayed, {"ETag": str(replayed["revision"]), "HUX-Replayed": "true", "Cache-Control": "no-store"})
exists = request.store.exists(FAMILY, record_id)
if expected != (request.store.get(FAMILY, record_id)["revision"] if exists else 0):
raise Conflict("mode selection revision does not match If-Match")
stamp = now_iso()
record = {
"id": record_id,
"schema": "hux.mode_selection.v1",
"owner": request.identity.subject,
"project_id": body["project_id"],
"conversation_id": conversation["id"],
"mode": mode,
"updated_at": stamp,
}
stored = request.store.put(FAMILY, record, expected_revision=expected)
request.store.append(FAMILY, "idempotency", {"key": key, "fingerprint": fingerprint, "id": record_id, "at": stamp})
with request.store.lock("conversations"):
current = request.store.get("conversations", conversation["id"])
request.store.put("conversations", {**current, "mode": mode["mode"], "updated_at": stamp}, current["revision"])
request.audit("modes.select", record_id)
return Response(200, stored, {"ETag": str(stored["revision"]), "Cache-Control": "no-store"})
def register(router: Router) -> None:
"""Attach HUX-06 routes."""
router.add("GET", "/hux/v1/modes", CARD, "modes.list", list_modes)
router.add("GET", "/hux/v1/projects/{project_id}/conversations/{id}/mode", CARD, "modes.read", get_selection)
router.add("PUT", "/hux/v1/projects/{project_id}/conversations/{id}/mode", CARD, "modes.select", put_selection)

View File

@ -0,0 +1,276 @@
"""HUX-07 scoped multimodal metadata, lineage and transcript corrections.
This service never accepts media bytes and has no capture endpoint. It only
records metadata after an approved autonomy action, plus inert camera/screen
intents that a human-facing client may send through the HUX-05 approval lane.
"""
from __future__ import annotations
import hashlib
import json
from typing import Any
from hux import contracts, redaction
from hux.errors import Conflict, Forbidden, Invalid, NotFound, TooLarge
from hux.http import Request, Response, Router, page
from hux.store import check_id, new_id, now_iso
CARD = "HUX-07"
ITEMS = "multimodal_items"
CORRECTIONS = "transcript_corrections"
INTENTS = "capture_intents"
MAX_ITEMS = 2000
MAX_CORRECTIONS = 10000
MAX_INTENTS = 1000
EXECUTABLE_MIMES = frozenset({"text/html", "application/xhtml+xml", "image/svg+xml", "application/xml", "text/xml"})
SAFE_SUFFIXES = (".jpg", ".jpeg", ".png", ".webp", ".wav", ".webm", ".ogg", ".mp4", ".pdf", ".txt")
KIND_MIMES = {
"image": frozenset({"image/jpeg", "image/png", "image/webp"}),
"audio": frozenset({"audio/wav", "audio/webm", "audio/ogg"}),
"video": frozenset({"video/webm", "video/mp4"}),
"document": frozenset({"application/pdf", "text/plain"}),
}
ITEM_FIELDS = frozenset({"project_id", "kind", "source", "filename", "mime", "bytes", "hash", "approval_id", "lineage"})
SCHEMAS = contracts.load_all()
SCHEMAS["multimodal.schema.json"] = contracts.load_schema("multimodal.schema.json")
def _body(request: Request, allowed: frozenset[str]) -> dict[str, Any]:
if not isinstance(request.body, dict):
raise Invalid("body must be a JSON object")
extra = set(request.body) - allowed
if extra:
raise Invalid("unexpected multimodal fields", sorted(extra))
return request.body
def _scope(request: Request, project_id: Any) -> dict[str, Any]:
from hux import organization
try:
conversation = request.store.get(organization.CONVERSATIONS, request.params["id"])
except (Invalid, NotFound) as error:
raise NotFound("project or conversation not found") from error
path_project = request.params.get("project_id")
actual = conversation.get("project_id")
if not isinstance(project_id, str) or project_id != path_project or not actual or project_id != actual:
raise NotFound("project or conversation not found")
if not organization.project_exists(request.store, project_id):
raise NotFound("project or conversation not found")
return conversation
def _if_match(request: Request, current: int) -> None:
expected = request.if_match()
if expected is None:
raise Invalid("If-Match is required")
if expected != current:
raise Conflict("revision does not match If-Match")
def _key(request: Request) -> str:
key = request.idempotency_key()
if not key:
raise Invalid("Idempotency-Key is required")
return key
def _digest(body: dict[str, Any]) -> str:
return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
def _replay(request: Request, family: str, key: str, digest: str) -> dict[str, Any] | None:
for row in request.store.read(family, "idempotency"):
if row.get("key") != key:
continue
if row.get("digest") != digest:
raise Conflict("Idempotency-Key was already used with different metadata")
return request.store.get(family, row["id"])
return None
def _remember(request: Request, family: str, key: str, digest: str, record_id: str) -> None:
request.store.append(family, "idempotency", {"key": key, "digest": digest, "id": record_id, "at": now_iso()})
def _validate(record: dict[str, Any], pointer: str) -> None:
problems = contracts.validate("multimodal.schema.json", record, SCHEMAS, pointer)
if problems:
raise Invalid("multimodal record failed contract validation", problems)
def _approval(request: Request, approval_id: Any, source: Any) -> str:
from hux import policy
try:
approval = policy.load_approval(request.store, check_id(approval_id))
except (Invalid, NotFound) as error:
raise Forbidden("approved autonomy action is required") from error
capability = "external_side_effect" if source in {"camera", "screen"} else "artifact_write"
if approval.get("status") != "approved" or approval.get("conversation_id") != request.params["id"]:
raise Forbidden("approved autonomy action is required")
if approval.get("capability") != capability:
raise Forbidden("approval does not cover this multimodal action")
return approval["id"]
def _lineage(request: Request, body: Any, project_id: str) -> dict[str, Any] | None:
if body is None:
return None
if not isinstance(body, dict) or set(body) - {"parent_item_id", "artifact_id", "artifact_version"}:
raise Invalid("lineage must contain only a parent item or artifact version")
if "parent_item_id" in body:
if len(body) != 1:
raise Invalid("parent-item lineage cannot also name an artifact")
parent = request.store.get(ITEMS, check_id(body["parent_item_id"]))
if parent["conversation_id"] != request.params["id"] or parent["project_id"] != project_id:
raise NotFound("lineage item not found")
return {"parent_item_id": parent["id"]}
if set(body) != {"artifact_id", "artifact_version"} or not isinstance(body["artifact_version"], int):
raise Invalid("artifact lineage needs artifact_id and artifact_version")
from hux import artifacts
artifact = request.store.get(artifacts.FAMILY, check_id(body["artifact_id"]))
if artifact.get("project_id") != project_id or artifact.get("conversation_id") != request.params["id"]:
raise NotFound("lineage artifact not found")
version = next((row for row in artifact["versions"] if row["version"] == body["artifact_version"]), None)
if version is None:
raise NotFound("lineage artifact version not found")
if version["content_ref"]["mime"].lower() in EXECUTABLE_MIMES or artifact.get("type") in {"html", "svg"}:
raise Invalid("executable HTML and SVG lineage is not accepted")
return {"artifact_id": artifact["id"], "artifact_version": version["version"]}
def create_item(request: Request) -> Response:
"""Register metadata for approved media; bytes are never accepted here."""
body = _body(request, ITEM_FIELDS)
conversation = _scope(request, body.get("project_id"))
key, digest = _key(request), _digest(body)
with request.store.lock(ITEMS):
replayed = _replay(request, ITEMS, key, digest)
if replayed is not None:
return Response(200, replayed, {"ETag": str(replayed["revision"]), "HUX-Replayed": "true", "Cache-Control": "no-store"})
_if_match(request, 0)
if request.store.count(ITEMS) >= MAX_ITEMS:
raise TooLarge("multimodal item limit reached")
filename, mime = body.get("filename"), body.get("mime")
if not isinstance(filename, str) or not filename.lower().endswith(SAFE_SUFFIXES):
raise Invalid("filename is missing or executable")
if not isinstance(mime, str) or mime.lower() in EXECUTABLE_MIMES:
raise Invalid("executable HTML, XML and SVG are not accepted")
if mime.lower() not in KIND_MIMES.get(body.get("kind"), frozenset()):
raise Invalid("kind and MIME type do not agree")
approval_id = _approval(request, body.get("approval_id"), body.get("source"))
lineage = _lineage(request, body.get("lineage"), body["project_id"])
record = {
"schema": "hux.multimodal_item.v1", "id": new_id("mmi"), "owner": request.identity.subject,
"project_id": body["project_id"], "conversation_id": conversation["id"], "kind": body.get("kind"),
"source": body.get("source"), "filename": filename, "mime": mime.lower(), "bytes": body.get("bytes"),
"hash": body.get("hash"), "approval_id": approval_id, "status": "metadata_only", "created_at": now_iso(),
}
if lineage:
record["lineage"] = lineage
_validate({**record, "revision": 1}, "/$defs/item")
stored = request.store.put(ITEMS, record, expected_revision=0)
_remember(request, ITEMS, key, digest, stored["id"])
request.audit("multimodal.create", stored["id"])
return Response(201, stored, {"ETag": "1", "Cache-Control": "no-store"})
def list_items(request: Request) -> Response:
"""List metadata in one project/conversation scope."""
project_id = request.params["project_id"]
_scope(request, project_id)
items = [row for row in request.store.scan(ITEMS) if row["project_id"] == project_id and row["conversation_id"] == request.params["id"]]
request.audit("multimodal.list", request.params["id"])
response = page(items)
response.headers["Cache-Control"] = "no-store"
return response
def get_item(request: Request) -> Response:
"""Read one metadata record only within its path scope."""
project_id = request.params["project_id"]
_scope(request, project_id)
try:
item = request.store.get(ITEMS, check_id(request.params["item_id"]))
except (Invalid, NotFound) as error:
raise NotFound("multimodal item not found") from error
if item["project_id"] != project_id or item["conversation_id"] != request.params["id"]:
raise NotFound("multimodal item not found")
request.audit("multimodal.read", item["id"])
return Response(200, item, {"ETag": str(item["revision"]), "Cache-Control": "no-store"})
def correct_transcript(request: Request) -> Response:
"""Append an immutable transcript correction and point the media head to it."""
body = _body(request, frozenset({"project_id", "replacement_text"}))
_scope(request, body.get("project_id"))
key, digest = _key(request), _digest({**body, "item_id": request.params["item_id"]})
with request.store.lock(CORRECTIONS):
replayed = _replay(request, CORRECTIONS, key, digest)
if replayed is not None:
return Response(200, replayed, {"HUX-Replayed": "true", "Cache-Control": "no-store"})
item = request.store.get(ITEMS, check_id(request.params["item_id"]))
if item["project_id"] != body["project_id"] or item["conversation_id"] != request.params["id"]:
raise NotFound("multimodal item not found")
if item["kind"] not in {"audio", "video"}:
raise Invalid("only audio or video transcripts can be corrected")
_if_match(request, item["revision"])
text = body.get("replacement_text")
if not isinstance(text, str) or not text.strip() or len(text) > 10000:
raise Invalid("replacement_text must contain 1 to 10000 characters")
if request.store.count(CORRECTIONS) >= MAX_CORRECTIONS:
raise TooLarge("transcript correction limit reached")
record = {
"schema": "hux.transcript_correction.v1", "id": new_id("trc"), "owner": request.identity.subject,
"project_id": body["project_id"], "conversation_id": request.params["id"], "item_id": item["id"],
"replacement_text": text, "created_at": now_iso(),
}
_validate({**record, "revision": 1}, "/$defs/correction")
stored = request.store.put(CORRECTIONS, record, expected_revision=0)
with request.store.lock(ITEMS):
updated = request.store.put(ITEMS, {**item, "latest_correction_id": stored["id"]}, item["revision"])
_remember(request, CORRECTIONS, key, digest, stored["id"])
request.audit("multimodal.correct", stored["id"])
return Response(201, stored, {"ETag": str(updated["revision"]), "Cache-Control": "no-store"})
def create_capture_intent(request: Request) -> Response:
"""Record an inert camera/screen proposal; it never grants execution."""
body = _body(request, frozenset({"project_id", "source", "purpose"}))
_scope(request, body.get("project_id"))
key, digest = _key(request), _digest(body)
with request.store.lock(INTENTS):
replayed = _replay(request, INTENTS, key, digest)
if replayed is not None:
return Response(200, replayed, {"HUX-Replayed": "true", "Cache-Control": "no-store"})
_if_match(request, 0)
if request.store.count(INTENTS) >= MAX_INTENTS:
raise TooLarge("capture intent limit reached")
purpose = body.get("purpose")
if body.get("source") not in {"camera", "screen"} or not isinstance(purpose, str) or not purpose.strip():
raise Invalid("source and purpose are required")
purpose = redaction.scrub_text(purpose[:280])[0]
record = {
"schema": "hux.capture_intent.v1", "id": new_id("cap"), "owner": request.identity.subject,
"project_id": body["project_id"], "conversation_id": request.params["id"], "source": body["source"],
"purpose": purpose, "status": "proposed", "requires_approval": True, "execution_allowed": False,
"created_at": now_iso(),
}
_validate({**record, "revision": 1}, "/$defs/capture_intent")
stored = request.store.put(INTENTS, record, expected_revision=0)
_remember(request, INTENTS, key, digest, stored["id"])
request.audit("multimodal.intent", stored["id"])
return Response(201, stored, {"ETag": "1", "Cache-Control": "no-store"})
def register(router: Router) -> None:
"""Attach metadata-only HUX-07 routes."""
base = "/hux/v1/projects/{project_id}/conversations/{id}"
router.add("POST", base + "/multimodal/items", CARD, "multimodal.create", create_item)
router.add("GET", base + "/multimodal/items", CARD, "multimodal.list", list_items)
router.add("GET", base + "/multimodal/items/{item_id}", CARD, "multimodal.read", get_item)
router.add("POST", base + "/multimodal/items/{item_id}/transcript-corrections", CARD, "multimodal.correct", correct_transcript)
router.add("POST", base + "/capture-intents", CARD, "multimodal.intent", create_capture_intent)

View File

@ -0,0 +1,333 @@
"""HUX-12 immutable, hash-chained deployment follow-through.
Each release is bound to one project and conversation. The current view is
reconstructed from append-only entries; every read verifies sequence, hash
chain and contract before returning anything. ``live_verified`` is reachable
only from a digest-converged pod with a passing health check.
"""
from __future__ import annotations
import hashlib
import json
import re
from typing import Any
from urllib.parse import urlsplit
from hux import contracts, redaction
from hux.errors import Conflict, Invalid, NotFound, TooLarge
from hux.http import Request, Response, Router, page
from hux.store import check_id, new_id, now_iso
CARD = "HUX-12"
FAMILY = "release_ledger"
MAX_RELEASES = 500
ZERO_HASH = "sha256:" + "0" * 64
ORDER = ("reviewed", "merged", "built", "verified", "deployed", "converged", "live_verified")
WORKLOADS = frozenset({"hermes-chat-router", "hermes-chat-tenant", "hermes-webui", "hermes-agent", "hermes-switchyard"})
EVIDENCE_FIELDS = frozenset({"review_url", "merge_commit", "ci_build_url", "image_ref", "image_digest", "harbor_digest", "flux_revision", "pod_digest", "health_check", "rollback_target"})
SHA = re.compile(r"^sha256:[0-9a-f]{64}$")
COMMIT = re.compile(r"^[0-9a-f]{40}$")
IMAGE = re.compile(r"^[a-z0-9./-]+:[A-Za-z0-9._-]+@(sha256:[0-9a-f]{64})$")
SCHEMAS = contracts.load_all()
SCHEMAS["release-ledger.schema.json"] = contracts.load_schema("release-ledger.schema.json")
def _body(request: Request, allowed: set[str]) -> dict[str, Any]:
if not isinstance(request.body, dict):
raise Invalid("body must be a JSON object")
extra = set(request.body) - allowed
if extra:
raise Invalid("unexpected release fields", sorted(extra))
return request.body
def _scope(request: Request) -> tuple[str, str]:
from hux import organization
project_id, conversation_id = request.params["project_id"], request.params["id"]
try:
conversation = request.store.get(organization.CONVERSATIONS, conversation_id)
except (Invalid, NotFound) as error:
raise NotFound("project or conversation not found") from error
if conversation.get("project_id") != project_id or not organization.project_exists(request.store, project_id):
raise NotFound("project or conversation not found")
return project_id, conversation_id
def _actor(request: Request) -> dict[str, str]:
if request.identity.trust == "worker":
return {"type": "system", "id": "release-worker"}
return {"type": "user", "id": request.identity.subject}
def _safe_url(value: Any, *, relative: bool = False) -> str:
if not isinstance(value, str) or not value or len(value) > 500:
raise Invalid("evidence URL is missing or too long")
parsed = urlsplit(value)
if parsed.query or parsed.fragment or parsed.username or parsed.password:
raise Invalid("evidence URLs may not contain credentials, queries or fragments")
if relative:
if not ((not parsed.scheme and value.startswith("/")) or parsed.scheme == "https"):
raise Invalid("health URL must be HTTPS or a local absolute path")
elif parsed.scheme != "https" or not parsed.netloc:
raise Invalid("evidence URL must use HTTPS")
if redaction.scrub_text(value)[1]:
raise Invalid("evidence URL appears to contain a secret")
return value
def _canonical(data: Any) -> bytes:
return json.dumps(data, sort_keys=True, separators=(",", ":")).encode()
def _entry_hash(entry: dict[str, Any]) -> str:
unsigned = {key: value for key, value in entry.items() if key != "entry_hash"}
return "sha256:" + hashlib.sha256(_canonical(unsigned)).hexdigest()
def _validate_release(release: dict[str, Any]) -> None:
problems = contracts.validate("release.schema.json", release, SCHEMAS)
if problems:
raise Invalid("release failed contract validation", problems)
def _validate_entry(entry: dict[str, Any]) -> None:
problems = contracts.validate("release-ledger.schema.json", entry, SCHEMAS)
if problems:
raise Invalid("release ledger entry failed contract validation", problems)
def _rows(request: Request, release_id: str) -> list[dict[str, Any]]:
try:
check_id(release_id)
except Invalid as error:
raise NotFound("release not found") from error
rows = request.store.read(FAMILY, release_id)
if not rows:
raise NotFound("release not found")
previous = ZERO_HASH
for sequence, row in enumerate(rows, 1):
if row.get("sequence") != sequence or row.get("previous_hash") != previous or row.get("entry_hash") != _entry_hash(row):
raise Invalid("release ledger integrity check failed")
_validate_entry(row)
previous = row["entry_hash"]
return rows
def _append(request: Request, release: dict[str, Any], project_id: str, conversation_id: str, previous: list[dict[str, Any]]) -> dict[str, Any]:
entry = {
"schema": "hux.release_ledger_entry.v1", "release_id": release["id"], "owner": request.identity.subject,
"project_id": project_id, "conversation_id": conversation_id, "sequence": len(previous) + 1,
"at": now_iso(), "previous_hash": previous[-1]["entry_hash"] if previous else ZERO_HASH, "snapshot": release,
}
entry["entry_hash"] = _entry_hash(entry)
_validate_entry(entry)
request.store.append(FAMILY, release["id"], entry)
return entry
def _view(entry: dict[str, Any]) -> dict[str, Any]:
return {
"release": entry["snapshot"],
"scope": {"project_id": entry["project_id"], "conversation_id": entry["conversation_id"]},
"revision": entry["sequence"],
"ledger_hash": entry["entry_hash"],
}
def _key(request: Request) -> str:
key = request.idempotency_key()
if not key:
raise Invalid("Idempotency-Key is required")
return key
def _fingerprint(body: dict[str, Any], path: str) -> str:
return hashlib.sha256(_canonical({"body": body, "path": path})).hexdigest()
def _replay(request: Request, key: str, fingerprint: str) -> dict[str, Any] | None:
for row in request.store.read(FAMILY, "idempotency"):
if row.get("key") != key:
continue
if row.get("fingerprint") != fingerprint:
raise Conflict("Idempotency-Key was already used for different release evidence")
rows = _rows(request, row["release_id"])
if row["sequence"] > len(rows):
raise Invalid("release idempotency ledger is inconsistent")
return _view(rows[row["sequence"] - 1])
return None
def _remember(request: Request, key: str, fingerprint: str, release_id: str, sequence: int) -> None:
request.store.append(FAMILY, "idempotency", {"key": key, "fingerprint": fingerprint, "release_id": release_id, "sequence": sequence, "at": now_iso()})
def _initial_evidence(body: dict[str, Any]) -> dict[str, Any]:
evidence = body.get("evidence")
if not isinstance(evidence, dict) or set(evidence) != {"review_url"}:
raise Invalid("reviewed release requires exactly evidence.review_url")
return {"review_url": _safe_url(evidence["review_url"])}
def create_release(request: Request) -> Response:
"""Create the sole reviewed ledger for one scoped workload commit."""
body = _body(request, {"workload", "commit", "feature_flags", "evidence"})
project_id, conversation_id = _scope(request)
key, fingerprint = _key(request), _fingerprint(body, request.path)
if request.if_match() != 0:
raise Invalid("If-Match: 0 is required when creating a release")
workload, commit = body.get("workload"), body.get("commit")
if workload not in WORKLOADS or not isinstance(commit, str) or not COMMIT.fullmatch(commit):
raise Invalid("workload and exact 40-character commit are required")
flags = body.get("feature_flags", [])
if not isinstance(flags, list) or len(flags) > 32:
raise Invalid("feature_flags must be a bounded list")
with request.store.lock(FAMILY):
replayed = _replay(request, key, fingerprint)
if replayed is not None:
return Response(200, replayed, {"HUX-Replayed": "true", "Cache-Control": "no-store"})
releases = [name for name in request.store.ledgers(FAMILY) if name.startswith("rel_")]
if len(releases) >= MAX_RELEASES:
raise TooLarge("release ledger limit reached")
for name in releases:
entry = _rows(request, name)[0]
release = entry["snapshot"]
if entry["project_id"] == project_id and entry["conversation_id"] == conversation_id and release["workload"] == workload and release["commit"] == commit:
raise Conflict("a release already exists for this scoped workload commit")
release = {
"schema": "hux.release.v1", "id": new_id("rel"), "workload": workload, "feature_flags": flags,
"commit": commit, "state": "reviewed", "evidence": _initial_evidence(body), "transitions": [],
}
_validate_release(release)
entry = _append(request, release, project_id, conversation_id, [])
result = _view(entry)
_remember(request, key, fingerprint, release["id"], 1)
request.audit("releases.create", release["id"])
return Response(201, result, {"ETag": "1", "Cache-Control": "no-store"})
def _transition_evidence(current: dict[str, Any], target: str, supplied: Any) -> dict[str, Any]:
if not isinstance(supplied, dict) or set(supplied) - EVIDENCE_FIELDS:
raise Invalid("evidence must be a bounded object of known fields")
expected = {
"merged": {"merge_commit"},
"built": {"ci_build_url", "image_ref", "image_digest", "harbor_digest"},
"verified": set(),
"deployed": {"flux_revision"},
"converged": {"pod_digest"},
"live_verified": {"health_check"},
"rolled_back": {"rollback_target"},
}[target]
if set(supplied) != expected:
raise Invalid(f"{target} requires exactly: {', '.join(sorted(expected)) or 'no new evidence'}")
evidence = {**current.get("evidence", {}), **supplied}
if target == "merged" and not COMMIT.fullmatch(str(evidence.get("merge_commit", ""))):
raise Invalid("merged requires an exact merge commit")
if target == "built":
match = IMAGE.fullmatch(str(evidence.get("image_ref", "")))
digest = evidence.get("image_digest")
if not match or not SHA.fullmatch(str(digest)) or evidence.get("harbor_digest") != digest or match.group(1) != digest:
raise Invalid("image ref, image digest and Harbor digest must match exactly")
evidence["ci_build_url"] = _safe_url(evidence.get("ci_build_url"))
if target == "verified" and evidence.get("harbor_digest") != evidence.get("image_digest"):
raise Invalid("verified image does not match Harbor")
if target == "deployed" and not re.fullmatch(r"main@sha1:[0-9a-f]{40}", str(evidence.get("flux_revision", ""))):
raise Invalid("deployed requires an exact Flux revision")
if target == "converged" and evidence.get("pod_digest") != evidence.get("image_digest"):
raise Invalid("pod digest does not equal the built image digest")
if target == "live_verified":
health = evidence.get("health_check")
if not isinstance(health, dict) or set(health) != {"url", "status", "at"} or health.get("status") != "pass":
raise Invalid("live verification requires one passing health check")
health["url"] = _safe_url(health["url"], relative=True)
if target == "rolled_back" and not SHA.fullmatch(str(evidence.get("rollback_target", ""))):
raise Invalid("rollback requires an exact target digest")
return evidence
def _evidence_refs(target: str, evidence: dict[str, Any]) -> list[dict[str, str]]:
mapping = {
"merged": ("build", "merge_commit"), "built": ("build", "image_digest"),
"verified": ("build", "harbor_digest"), "deployed": ("flux", "flux_revision"),
"converged": ("pod", "pod_digest"), "live_verified": ("receipt", "health_check"),
"rolled_back": ("pod", "rollback_target"),
}
kind, key = mapping[target]
value = evidence[key]
identifier = value["at"] if key == "health_check" else value
return [{"kind": kind, "id": str(identifier)[:200]}]
def transition_release(request: Request) -> Response:
"""Append exactly one legal, evidence-backed release transition."""
body = _body(request, {"to", "evidence"})
project_id, conversation_id = _scope(request)
key, fingerprint = _key(request), _fingerprint(body, request.path)
expected = request.if_match()
if expected is None:
raise Invalid("If-Match is required")
with request.store.lock(FAMILY):
replayed = _replay(request, key, fingerprint)
if replayed is not None:
return Response(200, replayed, {"HUX-Replayed": "true", "Cache-Control": "no-store"})
rows = _rows(request, request.params["release_id"])
head, current = rows[-1], rows[-1]["snapshot"]
if head["project_id"] != project_id or head["conversation_id"] != conversation_id:
raise NotFound("release not found")
if expected != len(rows):
raise Conflict("release revision does not match If-Match")
target = body.get("to")
if target == "rolled_back":
if current["state"] not in {"built", "verified", "deployed", "converged", "live_verified"}:
raise Conflict("there is no deployed image to roll back")
elif current["state"] == "rolled_back" or target not in ORDER or ORDER.index(target) != ORDER.index(current["state"]) + 1:
raise Conflict("release transition skips, reverses, or follows a terminal state")
evidence = _transition_evidence(current, target, body.get("evidence"))
stamp = now_iso()
transition = {"from": current["state"], "to": target, "at": stamp, "by": _actor(request), "evidence": _evidence_refs(target, evidence)}
release = {**current, "state": target, "evidence": evidence, "transitions": [*current["transitions"], transition]}
_validate_release(release)
entry = _append(request, release, project_id, conversation_id, rows)
result = _view(entry)
_remember(request, key, fingerprint, release["id"], entry["sequence"])
request.audit("releases.transition", release["id"], reason=target)
return Response(200, result, {"ETag": str(entry["sequence"]), "Cache-Control": "no-store"})
def get_release(request: Request) -> Response:
"""Read one verified release view within its project/conversation scope."""
project_id, conversation_id = _scope(request)
rows = _rows(request, request.params["release_id"])
head = rows[-1]
if head["project_id"] != project_id or head["conversation_id"] != conversation_id:
raise NotFound("release not found")
request.audit("releases.read", request.params["release_id"])
return Response(200, _view(head), {"ETag": str(len(rows)), "Cache-Control": "no-store"})
def list_releases(request: Request) -> Response:
"""List verified heads for one scope, newest ledger first."""
project_id, conversation_id = _scope(request)
items = []
for name in request.store.ledgers(FAMILY):
if not name.startswith("rel_"):
continue
head = _rows(request, name)[-1]
if head["project_id"] == project_id and head["conversation_id"] == conversation_id:
items.append(_view(head))
items.sort(key=lambda row: row["release"]["transitions"][-1]["at"] if row["release"]["transitions"] else "", reverse=True)
request.audit("releases.list", conversation_id)
response = page(items)
response.headers["Cache-Control"] = "no-store"
return response
def register(router: Router) -> None:
"""Attach HUX-12 routes."""
base = "/hux/v1/projects/{project_id}/conversations/{id}/releases"
router.add("POST", base, CARD, "releases.create", create_release)
router.add("GET", base, CARD, "releases.list", list_releases)
router.add("GET", base + "/{release_id}", CARD, "releases.read", get_release)
router.add("POST", base + "/{release_id}/transitions", CARD, "releases.transition", transition_release)

View File

@ -0,0 +1,138 @@
"""Bounded in-process HUX-10 retention scheduling for one tenant pod."""
from __future__ import annotations
import hashlib
import threading
from collections.abc import Callable, Mapping
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from hux import privacy
from hux.errors import Unauthorized
from hux.identity import SLOT_RE, Identity, _read_subject_binding
from hux.store import TenantStore
DAY_SECONDS = 86_400.0
MAX_JITTER_SECONDS = 3_600
MAX_BACKOFF_SECONDS = 900.0
class BindingUnavailable(ValueError):
"""The projected subject binding is absent or unsafe."""
def subject_binding(environ: Mapping[str, str]) -> Identity:
"""Read the canonical edge-published binding and combine it with the configured tenant slot."""
slot = environ.get("HUX_TENANT_SLOT", "")
raw_path = environ.get("HUX_SUBJECT_BINDING_FILE", "")
if not SLOT_RE.fullmatch(slot) or not raw_path:
raise BindingUnavailable("binding configuration unavailable")
try:
subject = _read_subject_binding(Path(raw_path))
except Unauthorized as error:
raise BindingUnavailable("binding file unavailable") from error
if subject is None:
raise BindingUnavailable("binding file unavailable")
return Identity(slot, subject, "worker", "worker")
class RetentionScheduler:
"""Run privacy retention without leaving the API process or tenant lock domain."""
def __init__(
self,
router: Any,
*,
clock: Callable[[], datetime] | None = None,
runner: Callable[..., dict[str, Any]] | None = None,
waiter: Callable[[float], bool] | None = None,
daily_seconds: float = DAY_SECONDS,
) -> None:
self.router = router
self.clock = clock or (lambda: datetime.now(timezone.utc))
self.runner = runner or privacy.run_retention
self.daily_seconds = max(1.0, daily_seconds)
self._stop = threading.Event()
self.waiter = waiter or self._stop.wait
self._thread: threading.Thread | None = None
self._lock = threading.Lock()
self._health: dict[str, Any] = {
"enabled": bool(router.flags.enabled("HUX-10")),
"status": "starting" if router.flags.enabled("HUX-10") else "disabled",
"last_run": None,
"last_success": None,
"errors": 0,
}
def health(self) -> dict[str, Any]:
"""Return bounded, credential-free scheduler evidence."""
with self._lock:
return dict(self._health)
def _set(self, **changes: Any) -> None:
with self._lock:
self._health.update(changes)
def _jitter(self, identity: Identity) -> float:
try:
cap = int(self.router.environ.get("HUX_RETENTION_JITTER_SECONDS", "900"))
except ValueError:
cap = 900
cap = min(MAX_JITTER_SECONDS, max(0, cap))
if cap == 0:
return 0.0
digest = hashlib.sha256(f"{identity.tenant_slot}\0{identity.subject}".encode()).digest()
return float(int.from_bytes(digest[:4], "big") % (cap + 1))
@staticmethod
def _backoff(failures: int) -> float:
return min(MAX_BACKOFF_SECONDS, float(5 * (2 ** min(max(failures - 1, 0), 7))))
def run(self) -> None:
"""Wait for a binding, run stale startup work, then run once per day."""
if not self.health()["enabled"]:
return
failures = 0
first = True
while not self._stop.is_set():
try:
identity = subject_binding(self.router.environ)
tenant = TenantStore(self.router.data_root, identity)
now = self.clock()
ran = not first or privacy.audit_stale(tenant, now)
if ran:
self.runner(tenant, now, identity)
stamp = now.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
self._set(last_run=stamp, last_success=stamp)
failures = 0
first = False
self._set(status="healthy", errors=0)
if self.waiter(self.daily_seconds + self._jitter(identity)):
break
except BindingUnavailable:
failures += 1
self._set(status="waiting_for_binding", errors=failures)
if self.waiter(self._backoff(failures)):
break
except Exception: # noqa: BLE001 - background failure cannot take down the API
failures += 1
self._set(status="error", errors=failures)
if self.waiter(self._backoff(failures)):
break
self._set(status="stopped")
def start(self) -> None:
"""Start one daemon thread when HUX-10 is enabled."""
if not self.health()["enabled"] or self._thread is not None:
return
self._thread = threading.Thread(target=self.run, name="hux-retention", daemon=True)
self._thread.start()
def stop(self) -> None:
"""Wake and join the scheduler during server shutdown."""
self._stop.set()
if self._thread is not None and self._thread is not threading.current_thread():
self._thread.join(timeout=2.0)
self._set(status="stopped")

View File

@ -13,7 +13,10 @@ from pathlib import Path
from hux.http import Router, serve
FAMILIES = ("foundation", "events", "memory", "privacy", "organization", "artifacts", "research", "policy")
FAMILIES = (
"foundation", "events", "memory", "privacy", "organization", "artifacts",
"modes", "research", "policy", "multimodal", "suggestions", "releases",
)
def build_router(data_root: Path, environ: dict[str, str] | None = None) -> Router:
@ -39,7 +42,10 @@ def main() -> None:
env = dict(os.environ)
router = build_router(Path(env.get("HUX_DATA_ROOT", "/opt/data")), env)
server = serve(router, bind_address(env), int(env.get("HUX_PORT", "8790")))
server.serve_forever()
try:
server.serve_forever()
finally:
server.server_close()
if __name__ == "__main__": # pragma: no cover

View File

@ -0,0 +1,237 @@
"""HUX-09 restrained contextual suggestions with server-owned suppression.
Clients submit a bounded context signal; they cannot create suggestion text,
change cooldowns, or claim that a suggestion was clicked. Private and
sensitive conversations return an explicit no-store result before any
suggestion state or idempotency record is written.
"""
from __future__ import annotations
import hashlib
import json
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from typing import Any
from hux import contracts
from hux.errors import Conflict, Invalid, NotFound
from hux.http import Request, Response, Router, page
from hux.store import now_iso
CARD = "HUX-09"
FAMILY = "suggestion_states"
MAX_STATES = 1000
CONTEXTS = ("first_session", "empty_project", "after_artifact", "after_research", "after_approval", "idle")
CATALOG: dict[str, dict[str, Any]] = {
"first_session": {"id": "sug_first_session", "kind": "tip", "title": "Choose how Hermes works", "body": "Pick Fast, Thoughtful, Research, Create, or Private/local for this conversation.", "action": {"type": "open_mode"}, "priority": 80},
"empty_project": {"id": "sug_empty_project", "kind": "project", "title": "Give this project a starting point", "body": "Add a goal or move a related conversation here when it is useful.", "action": {"type": "create_project"}, "priority": 50},
"after_artifact": {"id": "sug_after_artifact", "kind": "feature", "title": "Keep working on this artifact", "body": "Open the artifact workspace to compare versions or continue editing.", "action": {"type": "open_artifacts"}, "priority": 70},
"after_research": {"id": "sug_after_research", "kind": "workflow", "title": "Keep the research together", "body": "Save the sources, assumptions, and open questions as a reusable workflow.", "action": {"type": "start_workflow"}, "priority": 65},
"after_approval": {"id": "sug_after_approval", "kind": "tip", "title": "Review autonomy controls", "body": "You can adjust what Hermes asks before doing for this conversation.", "action": {"type": "none"}, "priority": 40},
"idle": {"id": "sug_idle_workflow", "kind": "workflow", "title": "Make repeated work reusable", "body": "If this repeats, Hermes can turn it into a workflow without running it now.", "action": {"type": "start_workflow"}, "priority": 20},
}
SCHEMAS = contracts.load_all()
clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc)
def _body(request: Request, allowed: set[str]) -> dict[str, Any]:
if not isinstance(request.body, dict):
raise Invalid("body must be a JSON object")
extra = set(request.body) - allowed
if extra:
raise Invalid("unexpected suggestion fields", sorted(extra))
return request.body
def _scope(request: Request) -> tuple[str, dict[str, Any]]:
from hux import organization
project_id, conversation_id = request.params["project_id"], request.params["id"]
try:
conversation = request.store.get(organization.CONVERSATIONS, conversation_id)
except (Invalid, NotFound) as error:
raise NotFound("project or conversation not found") from error
if conversation.get("project_id") != project_id or not organization.project_exists(request.store, project_id):
raise NotFound("project or conversation not found")
return project_id, conversation
def _privacy_gate(request: Request, conversation: dict[str, Any], body: dict[str, Any]) -> str | None:
from hux import privacy
if conversation.get("mode") == "private":
return "private_mode"
state = privacy.conversation_state(request.store, conversation["id"])
if state.get("forgotten") or state.get("memory_disabled"):
return "conversation_no_store"
if state.get("topics"):
return "sensitive_topic"
if body.get("no_store") is True or body.get("sensitivity") in {"sensitive", "restricted"}:
return "client_no_store"
return None
def _suggestion(context: str, surface: str) -> dict[str, Any]:
base = CATALOG[context]
record = {
"schema": "hux.suggestion.v1", **base,
"trigger": {"surface": surface, "context": context},
"suppression": {"dismissable": True, "max_shows": 3, "cooldown_seconds": 86400, "never_again_supported": True},
}
problems = contracts.validate("suggestion.schema.json", record, SCHEMAS, "/$defs/suggestion")
if problems:
raise Invalid("suggestion failed contract validation", problems)
return record
def _state_id(suggestion_id: str, project_id: str, conversation_id: str) -> str:
value = f"{suggestion_id}:{project_id}:{conversation_id}".encode()
return f"sugs_{hashlib.sha256(value).hexdigest()[:28]}"
def _public(state: dict[str, Any]) -> dict[str, Any]:
result = {key: state[key] for key in ("schema", "owner", "suggestion_id", "shows", "never_again")}
for key in ("last_shown_at", "dismissed_at", "acted_at"):
if key in state:
result[key] = state[key]
problems = contracts.validate("suggestion.schema.json", result, SCHEMAS, "/$defs/state")
if problems:
raise Invalid("suggestion state failed contract validation", problems)
return result
def _eligible(suggestion: dict[str, Any], state: dict[str, Any] | None, now: datetime) -> str | None:
if state is None:
return None
if state.get("never_again"):
return "never_again"
rules = suggestion["suppression"]
if state.get("shows", 0) >= rules["max_shows"]:
return "max_shows"
last = state.get("last_shown_at")
if last and now - datetime.fromisoformat(last.replace("Z", "+00:00")) < timedelta(seconds=rules["cooldown_seconds"]):
return "cooldown"
return None
def _key(request: Request) -> str:
key = request.idempotency_key()
if not key:
raise Invalid("Idempotency-Key is required")
return key
def _fingerprint(body: dict[str, Any], path: str) -> str:
return hashlib.sha256(json.dumps({"body": body, "path": path}, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
def _replay(request: Request, key: str, fingerprint: str) -> dict[str, Any] | None:
for row in request.store.read(FAMILY, "idempotency"):
if row.get("key") != key:
continue
if row.get("fingerprint") != fingerprint:
raise Conflict("Idempotency-Key was already used for a different suggestion action")
return row["result"]
return None
def evaluate(request: Request) -> Response:
"""Evaluate and atomically count one eligible suggestion display."""
body = _body(request, {"context", "no_store", "sensitivity"})
project_id, conversation = _scope(request)
context = body.get("context")
if context not in CONTEXTS:
raise Invalid("unknown suggestion context")
gated = _privacy_gate(request, conversation, body)
if gated:
request.audit("suggestions.evaluate", conversation["id"], reason=gated)
return Response(200, {"suggestion": None, "reason": gated, "stored": False}, {"Cache-Control": "no-store"})
key, fingerprint = _key(request), _fingerprint(body, request.path)
suggestion = _suggestion(context, request.identity.surface)
state_id = _state_id(suggestion["id"], project_id, conversation["id"])
with request.store.lock(FAMILY):
replayed = _replay(request, key, fingerprint)
if replayed is not None:
return Response(200, replayed, {"HUX-Replayed": "true", "Cache-Control": "no-store"})
state = request.store.get(FAMILY, state_id) if request.store.exists(FAMILY, state_id) else None
reason = _eligible(suggestion, state, clock())
if reason:
result = {"suggestion": None, "reason": reason, "stored": False}
else:
if state is None and request.store.count(FAMILY) >= MAX_STATES:
result = {"suggestion": None, "reason": "state_limit", "stored": False}
else:
stamp = now_iso()
record = {
"id": state_id, "schema": "hux.suggestion_state.v1", "owner": request.identity.subject,
"suggestion_id": suggestion["id"], "shows": (state or {}).get("shows", 0) + 1,
"last_shown_at": stamp, "never_again": False,
}
stored = request.store.put(FAMILY, record, expected_revision=state["revision"] if state else 0)
result = {"suggestion": suggestion, "state": _public(stored), "revision": stored["revision"], "stored": True}
request.store.append(FAMILY, "idempotency", {"key": key, "fingerprint": fingerprint, "result": result, "at": now_iso()})
request.audit("suggestions.evaluate", suggestion["id"], reason=result.get("reason", "shown"))
headers = {"Cache-Control": "no-store"}
if result.get("revision"):
headers["ETag"] = str(result["revision"])
return Response(200, result, headers)
def decide(request: Request) -> Response:
"""Record only an explicit user click, guarded by If-Match and idempotency."""
body = _body(request, {"decision", "clicked"})
_, conversation = _scope(request)
gated = _privacy_gate(request, conversation, body)
if gated:
raise NotFound("suggestion state not found")
if body.get("clicked") is not True or body.get("decision") not in {"dismissed", "acted", "never_again"}:
raise Invalid("an explicit clicked decision is required")
key, fingerprint = _key(request), _fingerprint(body, request.path)
expected = request.if_match()
if expected is None:
raise Invalid("If-Match is required")
state_id = _state_id(request.params["suggestion_id"], request.params["project_id"], conversation["id"])
with request.store.lock(FAMILY):
replayed = _replay(request, key, fingerprint)
if replayed is not None:
return Response(200, replayed, {"HUX-Replayed": "true", "Cache-Control": "no-store"})
if not request.store.exists(FAMILY, state_id):
raise NotFound("suggestion state not found")
current = request.store.get(FAMILY, state_id)
if current["revision"] != expected:
raise Conflict("suggestion state revision does not match If-Match")
stamp, decision = now_iso(), body["decision"]
updated = {**current}
if decision in {"dismissed", "never_again"}:
updated["dismissed_at"] = stamp
if decision == "acted":
updated["acted_at"] = stamp
if decision == "never_again":
updated["never_again"] = True
stored = request.store.put(FAMILY, updated, expected_revision=expected)
result = {"state": _public(stored), "revision": stored["revision"], "decision": decision}
request.store.append(FAMILY, "idempotency", {"key": key, "fingerprint": fingerprint, "result": result, "at": stamp})
request.audit("suggestions.decide", request.params["suggestion_id"], reason=decision)
return Response(200, result, {"ETag": str(stored["revision"]), "Cache-Control": "no-store"})
def list_states(request: Request) -> Response:
"""Inspect suppression state for exactly one project/conversation scope."""
project_id, conversation = _scope(request)
if _privacy_gate(request, conversation, {}):
return Response(200, {"items": [], "next": None}, {"Cache-Control": "no-store"})
prefix = lambda row: row["id"] == _state_id(row["suggestion_id"], project_id, conversation["id"])
items = [_public(row) | {"revision": row["revision"]} for row in request.store.scan(FAMILY) if prefix(row)]
request.audit("suggestions.list", conversation["id"])
response = page(items)
response.headers["Cache-Control"] = "no-store"
return response
def register(router: Router) -> None:
"""Attach HUX-09 routes."""
base = "/hux/v1/projects/{project_id}/conversations/{id}/suggestions"
router.add("POST", base + "/evaluate", CARD, "suggestions.evaluate", evaluate)
router.add("POST", base + "/{suggestion_id}/decisions", CARD, "suggestions.decide", decide)
router.add("GET", base + "/states", CARD, "suggestions.list", list_states)

View File

@ -0,0 +1,14 @@
{
"schema": "hux.capture_intent.v1",
"id": "cap_0001aaaa",
"owner": "usr_0123456789abcdef",
"project_id": "prj_0001aaaa",
"conversation_id": "conv_0001aaaa",
"source": "screen",
"purpose": "Show the application window relevant to this conversation.",
"status": "proposed",
"requires_approval": true,
"execution_allowed": false,
"created_at": "2026-08-24T12:02:00Z",
"revision": 1
}

View File

@ -0,0 +1,21 @@
{
"schema": "hux.context_bootstrap.v1",
"contract_version": "1.1.0",
"identity": {
"tenant_slot": "slot-3",
"subject": "usr_0123456789abcdef",
"surface": "worker",
"trust": "worker"
},
"session_id": "ses_0123456789abcdef0123456789abcdef",
"conversation_id": "conv_0123456789abcdef0123456789abcdef",
"project_id": "prj_0123456789abcdef0123456789abcdef",
"created": {
"project": true,
"conversation": true
},
"revisions": {
"project": 1,
"conversation": 1
}
}

View File

@ -0,0 +1,17 @@
{
"schema": "hux.multimodal_item.v1",
"id": "mmi_0001aaaa",
"owner": "usr_0123456789abcdef",
"project_id": "prj_0001aaaa",
"conversation_id": "conv_0001aaaa",
"kind": "audio",
"source": "voice",
"filename": "utterance.webm",
"mime": "audio/webm",
"bytes": 12000,
"hash": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"approval_id": "apr_0001aaaa",
"status": "metadata_only",
"created_at": "2026-08-24T12:00:00Z",
"revision": 1
}

View File

@ -0,0 +1,21 @@
{
"schema": "hux.release_ledger_entry.v1",
"release_id": "rel_0001aaaa",
"owner": "usr_0123456789abcdef",
"project_id": "prj_0001aaaa",
"conversation_id": "conv_0001aaaa",
"sequence": 1,
"at": "2026-08-24T12:03:00Z",
"previous_hash": "sha256:0000000000000000000000000000000000000000000000000000000000000000",
"entry_hash": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"snapshot": {
"schema": "hux.release.v1",
"id": "rel_0001aaaa",
"workload": "hermes-webui",
"feature_flags": ["hux.foundation"],
"commit": "1111111111111111111111111111111111111111",
"state": "reviewed",
"evidence": {"review_url": "https://git.bstein.dev/atlas/titan-iac/pulls/55"},
"transitions": []
}
}

View File

@ -0,0 +1,11 @@
{
"schema": "hux.transcript_correction.v1",
"id": "trc_0001aaaa",
"owner": "usr_0123456789abcdef",
"project_id": "prj_0001aaaa",
"conversation_id": "conv_0001aaaa",
"item_id": "mmi_0001aaaa",
"replacement_text": "This is the corrected transcript.",
"created_at": "2026-08-24T12:01:00Z",
"revision": 1
}

View File

@ -110,7 +110,8 @@
"frontend_owner": "codex",
"contracts": [
"artifact.schema.json",
"event.schema.json"
"event.schema.json",
"multimodal.schema.json"
],
"depends_on": [
"HUX-04",
@ -192,9 +193,12 @@
"backend_owner": "claude",
"frontend_owner": "codex",
"contracts": [
"release.schema.json"
"release.schema.json",
"release-ledger.schema.json"
],
"depends_on": [
"HUX-11"
],
"depends_on": [],
"default": false,
"rollback": "disable flag; release lane evidence archive remains the source of truth"
}

View File

@ -95,6 +95,58 @@
}
}
},
"context_bootstrap": {
"type": "object",
"additionalProperties": false,
"required": [
"schema",
"contract_version",
"identity",
"session_id",
"conversation_id",
"project_id",
"created",
"revisions"
],
"properties": {
"schema": {
"const": "hux.context_bootstrap.v1"
},
"contract_version": {
"$ref": "common.schema.json#/$defs/contract_version"
},
"identity": {
"$ref": "common.schema.json#/$defs/identity"
},
"session_id": {
"$ref": "common.schema.json#/$defs/id"
},
"conversation_id": {
"$ref": "common.schema.json#/$defs/id"
},
"project_id": {
"$ref": "common.schema.json#/$defs/id"
},
"created": {
"type": "object",
"additionalProperties": false,
"required": ["project", "conversation"],
"properties": {
"project": {"type": "boolean"},
"conversation": {"type": "boolean"}
}
},
"revisions": {
"type": "object",
"additionalProperties": false,
"required": ["project", "conversation"],
"properties": {
"project": {"$ref": "common.schema.json#/$defs/revision"},
"conversation": {"$ref": "common.schema.json#/$defs/revision"}
}
}
}
},
"error": {
"type": "object",
"additionalProperties": false,
@ -151,6 +203,9 @@
{
"$ref": "#/$defs/manifest"
},
{
"$ref": "#/$defs/context_bootstrap"
},
{
"$ref": "#/$defs/error"
}

View File

@ -0,0 +1,91 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://hermes.bstein.dev/contracts/hux/v1/multimodal.schema.json",
"title": "HUX scoped multimodal metadata",
"description": "Metadata-only records for HUX-07. Binary capture and upload are deliberately outside this API and remain gated by HUX-05.",
"$defs": {
"scope": {
"type": "object",
"additionalProperties": false,
"required": ["owner", "project_id", "conversation_id"],
"properties": {
"owner": {"$ref": "common.schema.json#/$defs/user_ref"},
"project_id": {"$ref": "common.schema.json#/$defs/id"},
"conversation_id": {"$ref": "common.schema.json#/$defs/id"}
}
},
"lineage": {
"type": "object",
"additionalProperties": false,
"properties": {
"parent_item_id": {"$ref": "common.schema.json#/$defs/id"},
"artifact_id": {"$ref": "common.schema.json#/$defs/id"},
"artifact_version": {"type": "integer", "minimum": 1}
}
},
"item": {
"type": "object",
"additionalProperties": false,
"required": ["schema", "id", "owner", "project_id", "conversation_id", "kind", "source", "filename", "mime", "bytes", "hash", "approval_id", "status", "created_at", "revision"],
"properties": {
"schema": {"const": "hux.multimodal_item.v1"},
"id": {"$ref": "common.schema.json#/$defs/id"},
"owner": {"$ref": "common.schema.json#/$defs/user_ref"},
"project_id": {"$ref": "common.schema.json#/$defs/id"},
"conversation_id": {"$ref": "common.schema.json#/$defs/id"},
"kind": {"type": "string", "enum": ["image", "audio", "video", "document"]},
"source": {"type": "string", "enum": ["upload", "camera", "screen", "voice"]},
"filename": {"type": "string", "minLength": 1, "maxLength": 160},
"mime": {"type": "string", "enum": ["image/jpeg", "image/png", "image/webp", "audio/wav", "audio/webm", "audio/ogg", "video/webm", "video/mp4", "application/pdf", "text/plain"]},
"bytes": {"type": "integer", "minimum": 1, "maximum": 26214400},
"hash": {"$ref": "common.schema.json#/$defs/sha256"},
"approval_id": {"$ref": "common.schema.json#/$defs/id"},
"lineage": {"$ref": "#/$defs/lineage"},
"latest_correction_id": {"$ref": "common.schema.json#/$defs/id"},
"status": {"const": "metadata_only"},
"created_at": {"$ref": "common.schema.json#/$defs/timestamp"},
"revision": {"$ref": "common.schema.json#/$defs/revision"}
}
},
"correction": {
"type": "object",
"additionalProperties": false,
"required": ["schema", "id", "owner", "project_id", "conversation_id", "item_id", "replacement_text", "created_at", "revision"],
"properties": {
"schema": {"const": "hux.transcript_correction.v1"},
"id": {"$ref": "common.schema.json#/$defs/id"},
"owner": {"$ref": "common.schema.json#/$defs/user_ref"},
"project_id": {"$ref": "common.schema.json#/$defs/id"},
"conversation_id": {"$ref": "common.schema.json#/$defs/id"},
"item_id": {"$ref": "common.schema.json#/$defs/id"},
"replacement_text": {"type": "string", "minLength": 1, "maxLength": 10000},
"created_at": {"$ref": "common.schema.json#/$defs/timestamp"},
"revision": {"$ref": "common.schema.json#/$defs/revision"}
}
},
"capture_intent": {
"type": "object",
"additionalProperties": false,
"required": ["schema", "id", "owner", "project_id", "conversation_id", "source", "purpose", "status", "requires_approval", "execution_allowed", "created_at", "revision"],
"properties": {
"schema": {"const": "hux.capture_intent.v1"},
"id": {"$ref": "common.schema.json#/$defs/id"},
"owner": {"$ref": "common.schema.json#/$defs/user_ref"},
"project_id": {"$ref": "common.schema.json#/$defs/id"},
"conversation_id": {"$ref": "common.schema.json#/$defs/id"},
"source": {"type": "string", "enum": ["camera", "screen"]},
"purpose": {"type": "string", "minLength": 1, "maxLength": 280},
"status": {"const": "proposed"},
"requires_approval": {"const": true},
"execution_allowed": {"const": false},
"created_at": {"$ref": "common.schema.json#/$defs/timestamp"},
"revision": {"$ref": "common.schema.json#/$defs/revision"}
}
}
},
"oneOf": [
{"$ref": "#/$defs/item"},
{"$ref": "#/$defs/correction"},
{"$ref": "#/$defs/capture_intent"}
]
}

View File

@ -0,0 +1,20 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://hermes.bstein.dev/contracts/hux/v1/release-ledger.schema.json",
"title": "HUX immutable release ledger entry",
"type": "object",
"additionalProperties": false,
"required": ["schema", "release_id", "owner", "project_id", "conversation_id", "sequence", "at", "previous_hash", "entry_hash", "snapshot"],
"properties": {
"schema": {"const": "hux.release_ledger_entry.v1"},
"release_id": {"$ref": "common.schema.json#/$defs/id"},
"owner": {"$ref": "common.schema.json#/$defs/user_ref"},
"project_id": {"$ref": "common.schema.json#/$defs/id"},
"conversation_id": {"$ref": "common.schema.json#/$defs/id"},
"sequence": {"type": "integer", "minimum": 1},
"at": {"$ref": "common.schema.json#/$defs/timestamp"},
"previous_hash": {"$ref": "common.schema.json#/$defs/sha256"},
"entry_hash": {"$ref": "common.schema.json#/$defs/sha256"},
"snapshot": {"$ref": "release.schema.json"}
}
}

View File

@ -0,0 +1,208 @@
"""HUX-11 authenticated deterministic context bootstrap contract."""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import contracts, foundation, identity, organization, store
from hux.server import build_router
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
SUBJECT = "usr_0123456789abcdef"
BASE = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": SUBJECT}
RELAY = {**BASE, "X-Hux-Surface": "telegram", "X-Hux-Trust": "relay", "X-Hux-Relay-Key": "rk"}
WORKER = {**BASE, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"}
SCHEMAS = contracts.load_all()
def context_key(tmp_path: Path, value: bytes = b"k" * 32, mode: int = 0o600) -> Path:
path = tmp_path / "context-key"
path.write_bytes(value)
path.chmod(mode)
return path
def router_for(tmp_path: Path, key_path: Path | None = None):
environ = {
"HUX_FLAGS": ALL_ON,
"HUX_RELAY_KEY": "rk",
"HUX_WORKER_KEY": "wk",
"HUX_ROUTER_KEY": "bk",
}
if key_path is not None:
environ["HUX_CONTEXT_KEY_FILE"] = str(key_path)
return build_router(tmp_path / "data", environ)
def payload(raw: str = "session-123", source: str = "home", who: identity.Identity | None = None) -> dict:
who = who or identity.Identity("slot-3", SUBJECT, "worker", "worker")
key = b"k" * 32
return {
"raw_session_id": raw,
"project_source": source,
"session_id": foundation.derive_context_id(key, "session", who, raw),
"conversation_id": foundation.derive_context_id(key, "conversation", who, raw),
"project_id": foundation.derive_context_id(key, "project", who, source),
}
def call(router, headers, body, idem: str = "bootstrap:0001"):
response = router.dispatch(
"POST", "/hux/v1/context/bootstrap", {**headers, "Idempotency-Key": idem}, json.dumps(body).encode()
)
return response.status, response.body
def tenant(router, subject: str = SUBJECT) -> store.TenantStore:
return store.TenantStore(router.data_root, identity.Identity("slot-3", subject, "worker", "worker"))
def test_relay_creates_and_worker_replays_without_persisting_sources(tmp_path):
router = router_for(tmp_path, context_key(tmp_path))
body = payload()
status, created = call(router, RELAY, body)
assert status == 201
assert created == {
"schema": "hux.context_bootstrap.v1",
"contract_version": "1.1.0",
"identity": {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "telegram", "trust": "relay"},
"session_id": body["session_id"],
"conversation_id": body["conversation_id"],
"project_id": body["project_id"],
"created": {"project": True, "conversation": True},
"revisions": {"project": 1, "conversation": 1},
}
assert contracts.validate_record(created, SCHEMAS) == []
status, replay = call(router, WORKER, body)
assert status == 200 and replay["created"] == {"project": False, "conversation": False}
assert replay["identity"]["trust"] == "worker"
written = " ".join(path.read_text() for path in (tmp_path / "data").rglob("*") if path.is_file())
assert body["raw_session_id"] not in written and body["project_source"] not in written
def test_bootstrap_makes_conversation_event_appendable(tmp_path):
router = router_for(tmp_path, context_key(tmp_path))
body = payload()
assert call(router, WORKER, body)[0] == 201
response = router.dispatch(
"POST",
f"/hux/v1/conversations/{body['conversation_id']}/events",
{**WORKER, "Idempotency-Key": "event:00000001"},
json.dumps({"kind": "message.user", "summary": "hello"}).encode(),
)
assert response.status == 201 and response.body["seq"] == 1
def test_only_relay_or_worker_may_bootstrap_and_identities_do_not_cross(tmp_path):
router = router_for(tmp_path, context_key(tmp_path))
browser = {**BASE, "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "bk"}
assert call(router, browser, payload())[0] == 403
other = identity.Identity("slot-3", "usr_fedcba9876543210", "worker", "worker")
headers = {**WORKER, "X-Hux-Subject": other.subject}
assert call(router, headers, payload(), "bootstrap:other")[0] == 400
assert call(router, headers, payload(who=other), "bootstrap:other2")[0] == 201
def test_id_proof_fields_and_idempotency_are_strict(tmp_path):
router = router_for(tmp_path, context_key(tmp_path))
good = payload()
for name, bad in (
("session_id", "ses_" + "0" * 32),
("conversation_id", "prj_" + "0" * 32),
("project_id", "prj_short"),
("raw_session_id", "x" * 241),
("project_source", "bad value"),
):
assert call(router, WORKER, {**good, name: bad}, f"bad:{name}:0001")[0] == 400
assert call(router, WORKER, good, "")[0] == 400
assert call(router, WORKER, {**good, "extra": True})[0] == 400
assert call(router, WORKER, good, "bootstrap:same")[0] == 201
assert call(router, WORKER, payload("session-456"), "bootstrap:same")[0] == 409
response = router.dispatch(
"POST", "/hux/v1/context/bootstrap", {**WORKER, "Idempotency-Key": "bootstrap:notobject"}, b"[]"
)
assert response.status == 400
def test_existing_owner_and_linkage_mismatches_are_never_overwritten(tmp_path):
router = router_for(tmp_path, context_key(tmp_path))
body = payload()
assert call(router, WORKER, body)[0] == 201
scoped = tenant(router)
conversation = scoped.get("conversations", body["conversation_id"])
scoped.put("conversations", {**conversation, "project_id": "prj_" + "f" * 32})
assert call(router, WORKER, body, "bootstrap:link2")[0] == 409
assert scoped.get("conversations", body["conversation_id"])["project_id"] == "prj_" + "f" * 32
binding = scoped.get(foundation.CONTEXT_FAMILY, body["session_id"])
scoped.put(foundation.CONTEXT_FAMILY, {**binding, "owner": "usr_fedcba9876543210"})
assert call(router, WORKER, body, "bootstrap:owner2")[0] == 409
def test_existing_project_owner_and_contract_damage_fail_closed(tmp_path):
router = router_for(tmp_path, context_key(tmp_path))
body = payload()
assert call(router, WORKER, body)[0] == 201
scoped = tenant(router)
project = scoped.get("projects", body["project_id"])
scoped.put("projects", {**project, "owner": "usr_fedcba9876543210"})
assert call(router, WORKER, body, "bootstrap:badowner")[0] == 409
scoped.put("projects", {**project, "name": ""})
assert call(router, WORKER, body, "bootstrap:badschema")[0] == 409
def test_bootstrap_honors_organization_caps_without_partial_creation(tmp_path, monkeypatch):
router = router_for(tmp_path, context_key(tmp_path))
monkeypatch.setattr(organization, "MAX_PROJECTS", 0)
assert call(router, WORKER, payload())[0] == 409
monkeypatch.setattr(organization, "MAX_PROJECTS", 200)
monkeypatch.setattr(organization, "MAX_CONVERSATIONS", 0)
assert call(router, WORKER, payload(), "bootstrap:convcap")[0] == 409
assert tenant(router).count("projects") == 0
@pytest.mark.parametrize("kind", ["missing", "mode", "length", "symlink"])
def test_context_key_must_be_exact_owner_regular_0600_file(tmp_path, kind):
key_path = tmp_path / "context-key"
if kind == "mode":
key_path = context_key(tmp_path, mode=0o640)
elif kind == "length":
key_path = context_key(tmp_path, b"short")
elif kind == "symlink":
target = context_key(tmp_path)
key_path = tmp_path / "context-link"
key_path.symlink_to(target)
router = router_for(tmp_path, None if kind == "missing" else key_path)
status, error = call(router, WORKER, payload())
assert status == 403 and error["code"] == "forbidden"
assert "kkkk" not in json.dumps(error)
def test_context_key_io_failures_are_sanitized(tmp_path, monkeypatch):
path = context_key(tmp_path)
monkeypatch.setattr(Path, "lstat", lambda self: (_ for _ in ()).throw(OSError("secret lstat detail")))
with pytest.raises(Exception, match="unavailable") as denied:
foundation._context_key({"HUX_CONTEXT_KEY_FILE": str(path)})
assert "secret lstat" not in str(denied.value)
monkeypatch.undo()
monkeypatch.setattr(Path, "read_bytes", lambda self: (_ for _ in ()).throw(OSError("secret read detail")))
with pytest.raises(Exception, match="unavailable") as denied:
foundation._context_key({"HUX_CONTEXT_KEY_FILE": str(path)})
assert "secret read" not in str(denied.value)
def test_foundation_capabilities_and_manifest_still_work(tmp_path):
router = router_for(tmp_path, context_key(tmp_path))
headers = {**BASE, "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "bk"}
capabilities = router.dispatch("GET", "/hux/v1/capabilities", headers, b"")
manifest = router.dispatch("GET", "/hux/v1/manifest", headers, b"")
assert capabilities.status == 200 and capabilities.body["schema"] == "hux.capabilities.v1"
assert manifest.status == 200 and manifest.body["schema"] == "hux.manifest.v1"

View File

@ -22,9 +22,9 @@ FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import audit, contracts, errors, flags, identity, store # noqa: E402
from hux.http import Router, Response, page, serve # noqa: E402
from hux.server import build_router # noqa: E402
from hux import audit, contracts, errors, flags, identity, store
from hux.http import Response, Router, page, serve
from hux.server import build_router
SCHEMAS = contracts.load_all()
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
@ -399,11 +399,14 @@ def test_server_main_wires_environment(tmp_path, monkeypatch):
def serve_forever(self):
seen["served"] = True
def server_close(self):
seen["closed"] = True
monkeypatch.setattr(server, "serve", lambda router, host, port: seen.update(root=router.data_root, host=host, port=port) or Fake())
monkeypatch.setenv("HUX_DATA_ROOT", str(tmp_path))
monkeypatch.setenv("HUX_PORT", "8791")
server.main()
assert seen == {"root": tmp_path, "host": "127.0.0.1", "port": 8791, "served": True}
assert seen == {"root": tmp_path, "host": "127.0.0.1", "port": 8791, "served": True, "closed": True}
# --- F2: worker allowlist, internal errors, health -----------------------------------
@ -434,12 +437,12 @@ def test_worker_trust_reaches_only_the_allowlisted_routes(tmp_path):
assert all(any(t == route.template for _, t in flags.WORKER_ROUTES if route.method == _) or (route.method, route.template) not in flags.WORKER_ROUTES for route in router.routes)
def test_unshipped_cards_declare_no_routes_and_cannot_enable():
"""F12: cards without a server implementation stay disabled even when every raw flag is configured."""
unshipped = {"HUX-06", "HUX-07", "HUX-09", "HUX-12"}
def test_new_backend_cards_declare_routes_and_enable_with_dependencies():
"""F12: each shipped backend card is route-backed and its full flag chain enables."""
shipped = {"HUX-06", "HUX-07", "HUX-09", "HUX-12"}
configured = flags.Flags({"HUX_FLAGS": ALL_ON})
assert all(flags.CARD_ROUTES[card] == [] for card in unshipped)
assert all(not configured.enabled(card) for card in unshipped)
assert all(flags.CARD_ROUTES[card] for card in shipped)
assert all(configured.enabled(card) for card in shipped)
def test_unexpected_handler_exceptions_become_a_500_error_record(tmp_path):
@ -464,7 +467,9 @@ def test_healthz_reports_the_contract_version(tmp_path):
try:
conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=5)
conn.request("GET", "/healthz")
assert json.loads(conn.getresponse().read()) == {"status": "ok", "contract_version": flags.CONTRACT_VERSION}
body = json.loads(conn.getresponse().read())
assert body["status"] == "ok" and body["contract_version"] == flags.CONTRACT_VERSION
assert body["retention"]["enabled"] is True
finally:
server.shutdown()
server.server_close()
@ -473,8 +478,8 @@ def test_healthz_reports_the_contract_version(tmp_path):
def test_no_outbound_network_client_in_the_service(tmp_path):
"""F13 / SO-29: no hux module imports urllib.request, requests, httpx or socket; http.server is the only http.* import."""
import re
banned = re.compile(r"^\s*(?:import|from)\s+(urllib\.request|requests|httpx|socket|http\.client)\b", re.M)
banned = re.compile(r"^\s*(?:import|from)\s+(urllib\.request|requests|httpx|socket|http\.client)\b", re.MULTILINE)
for path in sorted((FOUNDATION / "hux").glob("*.py")):
assert banned.search(path.read_text()) is None, path
http_imports = re.findall(r"^\s*from\s+(http\.\w+)\s+import", path.read_text(), re.M)
http_imports = re.findall(r"^\s*from\s+(http\.\w+)\s+import", path.read_text(), re.MULTILINE)
assert set(http_imports) <= {"http.server"}, (path, http_imports)

View File

@ -0,0 +1,83 @@
"""Registration and dependency contract for newly route-backed HUX families."""
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import contracts, flags, server
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
EXPECTED_FAMILIES = ("modes", "multimodal", "suggestions", "releases")
EXPECTED_ROUTES = {
"HUX-06": {
"/hux/v1/modes",
"/hux/v1/projects/{project_id}/conversations/{id}/mode",
},
"HUX-07": {
"/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items",
"/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items/{item_id}",
"/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items/{item_id}/transcript-corrections",
"/hux/v1/projects/{project_id}/conversations/{id}/capture-intents",
},
"HUX-09": {
"/hux/v1/projects/{project_id}/conversations/{id}/suggestions/evaluate",
"/hux/v1/projects/{project_id}/conversations/{id}/suggestions/{suggestion_id}/decisions",
"/hux/v1/projects/{project_id}/conversations/{id}/suggestions/states",
},
"HUX-12": {
"/hux/v1/projects/{project_id}/conversations/{id}/releases",
"/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}",
"/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}/transitions",
},
}
def test_server_registers_each_family_and_exact_card_routes(tmp_path):
assert all(name in server.FAMILIES for name in EXPECTED_FAMILIES)
router = server.build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
pairs = [(route.method, route.template) for route in router.routes]
assert len(pairs) == len(set(pairs)), "method/template registration must be unique"
registered = {route.template for route in router.routes}
for card, expected in EXPECTED_ROUTES.items():
assert set(flags.CARD_ROUTES[card]) == expected
assert expected <= registered
assert router.flags.enabled(card)
def test_release_followthrough_requires_foundation_but_not_activity():
registry = contracts.load_flags()
release = next(card for card in registry["cards"] if card["card"] == "HUX-12")
assert release["depends_on"] == ["HUX-11"]
assert release["contracts"] == ["release.schema.json", "release-ledger.schema.json"]
assert not flags.Flags({"HUX_FLAGS": "hux.release_followthrough"}).enabled("HUX-12")
enabled = flags.Flags({"HUX_FLAGS": "hux.foundation,hux.release_followthrough"})
assert enabled.enabled("HUX-12")
def test_contract_loader_and_worker_allowlist_are_complete():
schemas = contracts.load_all()
assert {"multimodal.schema.json", "release-ledger.schema.json"} <= schemas.keys()
multimodal = next(card for card in contracts.load_flags()["cards"] if card["card"] == "HUX-07")
assert multimodal["contracts"][-1] == "multimodal.schema.json"
worker_expected = {
("GET", "/hux/v1/modes"),
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/mode"),
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items"),
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}"),
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}/transitions"),
}
assert worker_expected <= flags.WORKER_ROUTES
assert ("PUT", "/hux/v1/projects/{project_id}/conversations/{id}/mode") not in flags.WORKER_ROUTES
assert not any("suggestions" in path for _, path in flags.WORKER_ROUTES)
def test_registration_test_stays_bounded():
assert len(Path(__file__).read_text().splitlines()) <= 500

View File

@ -0,0 +1,151 @@
"""Focused HUX-06 backend tests."""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import contracts, identity, modes, store
from hux.errors import Invalid
from hux.server import build_router
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
def call(router, method, path, body=None, headers=None):
raw = b"" if body is None else json.dumps(body).encode()
response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, raw)
return response.status, response.body, response.headers
@pytest.fixture
def setup(tmp_path, monkeypatch):
monkeypatch.setenv("HUX_SWITCHYARD_ROUTE_CATALOG", "atlas/manual/codex/gpt-5,atlas/manual/claude/opus,atlas/manual/local/qwen-14b")
router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
_, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "P"})
_, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
base = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}/mode"
return router, project, conversation, base
def test_catalog_is_provider_neutral_and_private_is_local(setup):
router, _, _, _ = setup
status, body, _ = call(router, "GET", "/hux/v1/modes")
assert status == 200 and [row["mode"] for row in body["items"]] == list(modes.MODE_NAMES)
for row in body["items"]:
assert contracts.validate("mode.schema.json", row) == []
if row["mode"] != "private":
assert row["switchyard"]["route_id"].startswith("atlas/auto/")
assert "/codex/" not in row["switchyard"]["route_id"] and "/claude/" not in row["switchyard"]["route_id"]
private = body["items"][-1]
assert private["constraints"]["providers"] == ["local"] and private["constraints"]["local_only"] is True
def test_select_get_replay_and_conversation_privacy_mode(setup):
router, project, conversation, base = setup
body = {"project_id": project["id"], "mode": "private"}
status, selected, headers = call(router, "PUT", base, body, {"If-Match": "0", "Idempotency-Key": "mode-key-0001"})
assert (status, selected["revision"], headers["ETag"]) == (200, 1, "1")
assert selected["mode"]["mode"] == "private" and selected["mode"]["switchyard"]["route_id"].startswith("atlas/manual/local/")
assert call(router, "GET", base)[1] == selected
stored_conversation = call(router, "GET", f"/hux/v1/conversations/{conversation['id']}")[1]
assert stored_conversation["mode"] == "private"
status, replay, replay_headers = call(router, "PUT", base, body, {"If-Match": "0", "Idempotency-Key": "mode-key-0001"})
assert status == 200 and replay == selected and replay_headers["HUX-Replayed"] == "true"
status, error, _ = call(router, "PUT", base, {**body, "mode": "fast"}, {"If-Match": "1", "Idempotency-Key": "mode-key-0001"})
assert (status, error["code"]) == (409, "conflict")
def test_advanced_pin_is_explicit_catalogued_and_constrained(setup):
router, project, _, base = setup
common = {"project_id": project["id"], "mode": "thoughtful"}
pin = "atlas/manual/claude/opus"
status, selected, _ = call(router, "PUT", base, {**common, "advanced": True, "override_route_id": pin}, {"If-Match": "0", "Idempotency-Key": "mode-key-0002"})
assert status == 200 and selected["mode"]["switchyard"]["override_route_id"] == pin
next_headers = {"If-Match": "1", "Idempotency-Key": "mode-key-0003"}
bad = [
{**common, "override_route_id": pin},
{**common, "advanced": True},
{**common, "advanced": True, "override_route_id": "atlas/manual/claude/not-catalogued"},
{**common, "advanced": True, "override_route_id": "atlas/auto/deep"},
{"project_id": project["id"], "mode": "private", "advanced": True, "override_route_id": "atlas/manual/claude/opus"},
]
for index, body in enumerate(bad):
status, error, _ = call(router, "PUT", base, body, {**next_headers, "Idempotency-Key": f"bad-mode-{index:03}"})
assert status == 400 and error["code"] == "invalid"
status, local, _ = call(router, "PUT", base, {"project_id": project["id"], "mode": "private", "advanced": True, "override_route_id": "atlas/manual/local/qwen-14b"}, next_headers)
assert status == 200 and local["revision"] == 2
@pytest.mark.parametrize("body,headers", [
({"project_id": None, "mode": "fast"}, {"If-Match": "0", "Idempotency-Key": "mode-key-0100"}),
({"project_id": "prj_wrong0000", "mode": "fast"}, {"If-Match": "0", "Idempotency-Key": "mode-key-0101"}),
({"mode": "fast"}, {"If-Match": "0"}),
({"mode": "fast"}, {"Idempotency-Key": "mode-key-0102"}),
({"mode": "unknown"}, {"If-Match": "0", "Idempotency-Key": "mode-key-0103"}),
])
def test_selection_fails_closed(setup, body, headers):
router, project, _, base = setup
if "project_id" not in body:
body["project_id"] = project["id"]
status, _, _ = call(router, "PUT", base, body, headers)
assert status in {400, 404}
def test_revision_scope_and_catalog_bounds(setup, monkeypatch):
router, project, _, base = setup
body = {"project_id": project["id"], "mode": "fast"}
assert call(router, "PUT", base, body, {"If-Match": "1", "Idempotency-Key": "mode-key-0200"})[0] == 409
assert call(router, "GET", base)[0] == 404
assert call(router, "GET", base, headers=OTHER)[0] == 404
wrong = base.replace(project["id"], "prj_missing0000")
assert call(router, "GET", wrong)[0] == 404
monkeypatch.setenv("HUX_SWITCHYARD_ROUTE_CATALOG", ",".join(f"atlas/manual/codex/r{i}" for i in range(257)))
advanced = {**body, "advanced": True, "override_route_id": "atlas/manual/codex/r1"}
assert call(router, "PUT", base, advanced, {"If-Match": "0", "Idempotency-Key": "mode-key-0201"})[0] == 400
monkeypatch.setenv("HUX_SWITCHYARD_ROUTE_CATALOG", "garbage,atlas/manual/codex/gpt-5")
assert modes._catalog() == {"atlas/manual/codex/gpt-5"}
def test_body_and_header_validation(setup):
router, project, _, base = setup
assert router.dispatch("PUT", base, HEADERS, b"[]").status == 400
body = {"project_id": project["id"], "mode": "fast"}
assert call(router, "PUT", base, body, {"If-Match": "x", "Idempotency-Key": "mode-key-0300"})[0] == 400
assert call(router, "PUT", base, body, {"If-Match": "0", "Idempotency-Key": "short"})[0] == 400
assert call(router, "PUT", base, {**body, "provider": "claude"}, {"If-Match": "0", "Idempotency-Key": "mode-key-0301"})[0] == 400
def test_missing_project_and_defensive_contract_failures(setup, monkeypatch):
router, project, _, base = setup
scoped = store.TenantStore(router.data_root, identity.resolve(HEADERS, {"HUX_ROUTER_KEY": "rk"}))
scoped.delete("projects", project["id"])
assert call(router, "GET", base)[0] == 404
original = modes.rules.mode_contract
monkeypatch.setattr(modes.rules, "mode_contract", lambda *_: (_ for _ in ()).throw(ValueError("bad mapping")))
with pytest.raises(Invalid, match="bad mapping"):
modes._contract("fast", False, None)
monkeypatch.setattr(modes.rules, "mode_contract", lambda *_: {"schema": "broken"})
with pytest.raises(Invalid, match="contract validation"):
modes._contract("fast", False, None)
monkeypatch.setattr(modes.rules, "mode_contract", original)
record = original("fast")
record["switchyard"]["route_id"] = "atlas/manual/codex/gpt-5"
monkeypatch.setattr(modes.rules, "mode_contract", lambda *_: record)
with pytest.raises(Invalid, match="may not pin"):
modes._contract("fast", False, None)
def test_source_and_test_files_stay_bounded():
assert len((FOUNDATION / "hux" / "modes.py").read_text().splitlines()) <= 500
assert len(Path(__file__).read_text().splitlines()) <= 500

View File

@ -0,0 +1,219 @@
"""Focused HUX-07 metadata-only backend tests."""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import contracts, identity, multimodal, store
from hux.server import build_router
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
SCHEMAS = contracts.load_all()
SCHEMAS["multimodal.schema.json"] = contracts.load_schema("multimodal.schema.json")
def call(router, method, path, body=None, headers=None):
raw = b"" if body is None else json.dumps(body).encode()
response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, raw)
return response.status, response.body, response.headers
def tenant(router, headers=HEADERS):
return store.TenantStore(router.data_root, identity.resolve(headers, {"HUX_ROUTER_KEY": "rk"}))
def approval(router, conversation_id, capability="artifact_write", name="apr_upload0001"):
return tenant(router).put("approvals", {
"id": name, "status": "approved", "conversation_id": conversation_id, "capability": capability,
})
@pytest.fixture
def setup(tmp_path):
router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
_, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "P"})
_, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
base = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}"
approval(router, conversation["id"])
return router, project, conversation, base
def item_body(project, **changes):
body = {
"project_id": project["id"], "kind": "audio", "source": "upload", "filename": "note.webm",
"mime": "audio/webm", "bytes": 1234, "hash": "sha256:" + "a" * 64, "approval_id": "apr_upload0001",
}
return {**body, **changes}
def create_item(setup, key="media-key-0001", **changes):
router, project, _, base = setup
return call(router, "POST", base + "/multimodal/items", item_body(project, **changes), {"If-Match": "0", "Idempotency-Key": key})
def test_metadata_create_list_get_and_replay(setup):
router, _, _, base = setup
status, item, headers = create_item(setup)
assert (status, item["status"], headers["ETag"]) == (201, "metadata_only", "1")
assert contracts.validate("multimodal.schema.json", item, SCHEMAS, "/$defs/item") == []
status, listed, headers = call(router, "GET", base + "/multimodal/items")
assert status == 200 and listed["items"] == [item] and headers["Cache-Control"] == "no-store"
status, got, headers = call(router, "GET", base + f"/multimodal/items/{item['id']}")
assert status == 200 and got == item and headers["ETag"] == "1"
status, replayed, headers = create_item(setup)
assert status == 200 and replayed == item and headers["HUX-Replayed"] == "true"
status, error, _ = create_item(setup, filename="different.webm")
assert (status, error["code"]) == (409, "conflict")
assert not list(Path(router.data_root).rglob("*.webm")), "the endpoint stores metadata, never media"
@pytest.mark.parametrize("changes", [
{"filename": "payload.html", "mime": "text/html", "kind": "document"},
{"filename": "drawing.svg", "mime": "image/svg+xml", "kind": "image"},
{"filename": "photo.png", "mime": "audio/webm", "kind": "image"},
{"filename": "photo.exe", "mime": "image/png", "kind": "image"},
{"bytes": 0}, {"bytes": 30_000_000}, {"hash": "sha256:no"}, {"kind": "binary"},
{"content": "raw bytes are forbidden"}, {"url": "https://example.com/a"},
])
def test_media_validation_rejects_executable_content_and_unbounded_metadata(setup, changes):
status, error, _ = create_item(setup, key=f"invalid-media-{abs(hash(repr(changes)))}", **changes)
assert status == 400 and error["code"] == "invalid"
def test_approval_must_cover_scope_and_action(setup):
router, project, conversation, base = setup
common = item_body(project)
approval(router, conversation["id"], "external_side_effect", "apr_capture0001")
cases = [
({**common, "approval_id": "apr_missing0001"}, 403),
({**common, "approval_id": "apr_capture0001"}, 403),
({**common, "source": "camera", "approval_id": "apr_upload0001"}, 403),
]
for index, (body, expected) in enumerate(cases):
status, _, _ = call(router, "POST", base + "/multimodal/items", body, {"If-Match": "0", "Idempotency-Key": f"approval-bad-{index}"})
assert status == expected
camera = {**common, "source": "camera", "approval_id": "apr_capture0001"}
assert call(router, "POST", base + "/multimodal/items", camera, {"If-Match": "0", "Idempotency-Key": "approval-good-1"})[0] == 201
other_conv = call(router, "POST", "/hux/v1/conversations", {"title": "Other", "project_id": project["id"]})[1]
approval(router, other_conv["id"], name="apr_other00001")
assert call(router, "POST", base + "/multimodal/items", {**common, "approval_id": "apr_other00001"}, {"If-Match": "0", "Idempotency-Key": "approval-bad-other"})[0] == 403
def test_parent_and_artifact_lineage_are_scope_checked(setup):
router, project, _, base = setup
parent = create_item(setup, key="lineage-parent-1")[1]
status, child, _ = create_item(setup, key="lineage-child-01", lineage={"parent_item_id": parent["id"]})
assert status == 201 and child["lineage"] == {"parent_item_id": parent["id"]}
artifact_body = {
"type": "document", "title": "Notes", "project_id": project["id"], "conversation_id": base.split("/")[-1],
"content": "plain", "mime": "text/plain",
}
artifact = call(router, "POST", "/hux/v1/artifacts", artifact_body, {"Idempotency-Key": "artifact-lineage-1"})[1]
lineage = {"artifact_id": artifact["id"], "artifact_version": 1}
status, linked, _ = create_item(setup, key="lineage-artifact1", lineage=lineage)
assert status == 201 and linked["lineage"] == lineage
bad = [
{"parent_item_id": parent["id"], "artifact_id": artifact["id"]},
{"artifact_id": artifact["id"]},
{"artifact_id": artifact["id"], "artifact_version": 99},
{"unknown": "mmi_missing000"},
]
for index, value in enumerate(bad):
assert create_item(setup, key=f"bad-lineage-{index}", lineage=value)[0] in {400, 404}
html = call(router, "POST", "/hux/v1/artifacts", {**artifact_body, "type": "html", "title": "Unsafe", "content": "<b>x</b>", "mime": "text/html"}, {"Idempotency-Key": "artifact-lineage-2"})[1]
assert create_item(setup, key="bad-lineage-html", lineage={"artifact_id": html["id"], "artifact_version": 1})[0] == 400
def test_transcript_corrections_are_immutable_and_concurrent(setup):
router, project, _, base = setup
item = create_item(setup)[1]
path = base + f"/multimodal/items/{item['id']}/transcript-corrections"
body = {"project_id": project["id"], "replacement_text": "What I actually said."}
status, correction, headers = call(router, "POST", path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-001"})
assert status == 201 and headers["ETag"] == "2"
assert contracts.validate("multimodal.schema.json", correction, SCHEMAS, "/$defs/correction") == []
got = call(router, "GET", base + f"/multimodal/items/{item['id']}")[1]
assert got["latest_correction_id"] == correction["id"] and got["revision"] == 2
status, replay, headers = call(router, "POST", path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-001"})
assert status == 200 and replay == correction and headers["HUX-Replayed"] == "true"
assert call(router, "POST", path, {**body, "replacement_text": "changed"}, {"If-Match": "2", "Idempotency-Key": "correct-key-001"})[0] == 409
assert call(router, "POST", path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-002"})[0] == 409
assert call(router, "POST", path, {**body, "replacement_text": ""}, {"If-Match": "2", "Idempotency-Key": "correct-key-003"})[0] == 400
image = create_item(setup, key="image-for-correct", kind="image", filename="a.png", mime="image/png")[1]
image_path = base + f"/multimodal/items/{image['id']}/transcript-corrections"
assert call(router, "POST", image_path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-004"})[0] == 400
def test_capture_intent_is_inert_redacted_and_idempotent(setup):
router, project, _, base = setup
token = "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
body = {"project_id": project["id"], "source": "screen", "purpose": f"Show issue {token}"}
headers = {"If-Match": "0", "Idempotency-Key": "capture-key-001"}
status, intent, response_headers = call(router, "POST", base + "/capture-intents", body, headers)
assert status == 201 and intent["status"] == "proposed" and intent["requires_approval"] is True
assert intent["execution_allowed"] is False and token not in intent["purpose"]
assert contracts.validate("multimodal.schema.json", intent, SCHEMAS, "/$defs/capture_intent") == []
status, replay, replay_headers = call(router, "POST", base + "/capture-intents", body, headers)
assert status == 200 and replay == intent and replay_headers["HUX-Replayed"] == "true"
for bad in ({**body, "source": "microphone"}, {**body, "purpose": ""}, {**body, "execute": True}):
assert call(router, "POST", base + "/capture-intents", bad, {"If-Match": "0", "Idempotency-Key": f"bad-cap-{abs(hash(repr(bad)))}"})[0] == 400
assert response_headers["Cache-Control"] == "no-store"
def test_scope_limits_and_malformed_requests_fail_closed(setup, monkeypatch):
router, project, _, base = setup
assert call(router, "GET", base + "/multimodal/items", headers=OTHER)[0] == 404
wrong = base.replace(project["id"], "prj_missing0000")
assert call(router, "GET", wrong + "/multimodal/items")[0] == 404
assert router.dispatch("POST", base + "/multimodal/items", HEADERS, b"[]").status == 400
body = item_body(project, filename="b.webm")
assert call(router, "POST", base + "/multimodal/items", body, {"Idempotency-Key": "missing-ifmatch"})[0] == 400
monkeypatch.setattr(multimodal, "MAX_ITEMS", 0)
assert create_item(setup, key="over-item-limit", filename="c.webm")[0] == 413
monkeypatch.setattr(multimodal, "MAX_INTENTS", 0)
intent = {"project_id": project["id"], "source": "camera", "purpose": "scan"}
assert call(router, "POST", base + "/capture-intents", intent, {"If-Match": "0", "Idempotency-Key": "over-intent-limit"})[0] == 413
assert call(router, "GET", base + "/multimodal/items/mmi_missing000")[0] == 404
def test_cross_scope_branches_missing_keys_and_correction_limit(setup, monkeypatch):
router, project, _, base = setup
assert call(router, "POST", base + "/multimodal/items", item_body(project), {"If-Match": "0"})[0] == 400
executable = item_body(project, filename="payload.txt", mime="text/html", kind="document")
assert call(router, "POST", base + "/multimodal/items", executable, {"If-Match": "0", "Idempotency-Key": "mime-executable1"})[0] == 400
_, other_conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "Other", "project_id": project["id"]})
other_base = f"/hux/v1/projects/{project['id']}/conversations/{other_conversation['id']}"
approval(router, other_conversation["id"], name="apr_othermedia1")
other_body = item_body(project, approval_id="apr_othermedia1")
other_item = call(router, "POST", other_base + "/multimodal/items", other_body, {"If-Match": "0", "Idempotency-Key": "other-media-key1"})[1]
assert create_item(setup, key="cross-parent-key1", lineage={"parent_item_id": other_item["id"]})[0] == 404
artifact = call(router, "POST", "/hux/v1/artifacts", {
"type": "document", "title": "Other", "project_id": project["id"], "conversation_id": other_conversation["id"],
"content": "plain", "mime": "text/plain",
}, {"Idempotency-Key": "other-artifact-key"})[1]
assert create_item(setup, key="cross-artifact-1", lineage={"artifact_id": artifact["id"], "artifact_version": 1})[0] == 404
mine = create_item(setup, key="my-cross-media1")[1]
assert call(router, "GET", other_base + f"/multimodal/items/{mine['id']}")[0] == 404
correction = {"project_id": project["id"], "replacement_text": "correct"}
assert call(router, "POST", other_base + f"/multimodal/items/{mine['id']}/transcript-corrections", correction, {"If-Match": "1", "Idempotency-Key": "cross-correct-1"})[0] == 404
monkeypatch.setattr(multimodal, "MAX_CORRECTIONS", 0)
assert call(router, "POST", base + f"/multimodal/items/{mine['id']}/transcript-corrections", correction, {"If-Match": "1", "Idempotency-Key": "correct-limit-1"})[0] == 413
scoped = tenant(router)
scoped.delete("projects", project["id"])
assert call(router, "GET", base + "/multimodal/items")[0] == 404
def test_source_contract_and_test_files_stay_bounded():
for path in (FOUNDATION / "hux" / "multimodal.py", ROOT / "services" / "hermes" / "contracts" / "hux" / "multimodal.schema.json", Path(__file__)):
assert len(path.read_text().splitlines()) <= 500

View File

@ -0,0 +1,272 @@
"""Focused HUX-12 immutable release-ledger tests."""
from __future__ import annotations
import json
import sys
from pathlib import Path
from types import SimpleNamespace
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import contracts, identity, releases, store
from hux.errors import Invalid
from hux.server import build_router
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
DIGEST = "sha256:" + "a" * 64
OTHER_DIGEST = "sha256:" + "b" * 64
SCHEMAS = contracts.load_all()
SCHEMAS["release-ledger.schema.json"] = contracts.load_schema("release-ledger.schema.json")
def call(router, method, path, body=None, headers=None):
raw = b"" if body is None else json.dumps(body).encode()
response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, raw)
return response.status, response.body, response.headers
def tenant(router):
return store.TenantStore(router.data_root, identity.resolve(HEADERS, {"HUX_ROUTER_KEY": "rk"}))
@pytest.fixture
def setup(tmp_path):
router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
_, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "P"})
_, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
base = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}/releases"
return router, project, conversation, base
def create(setup, key="release-key-001", commit="1" * 40, **changes):
router, _, _, base = setup
body = {
"workload": "hermes-webui", "commit": commit, "feature_flags": ["hux.foundation"],
"evidence": {"review_url": "https://git.bstein.dev/atlas/titan-iac/pulls/55"}, **changes,
}
return call(router, "POST", base, body, {"If-Match": "0", "Idempotency-Key": key})
def move(router, path, revision, key, target, evidence):
return call(router, "POST", path + "/transitions", {"to": target, "evidence": evidence}, {"If-Match": str(revision), "Idempotency-Key": key})
def build_evidence(digest=DIGEST):
return {
"ci_build_url": "https://jenkins.bstein.dev/job/hermes/20/",
"image_ref": f"registry.bstein.dev/bstein/hermes:build-20@{digest}",
"image_digest": digest,
"harbor_digest": digest,
}
def through_built(setup, key_suffix="a", commit="2" * 40):
router, _, _, base = setup
created = create(setup, f"create-built-{key_suffix}", commit)[1]
path = base + "/" + created["release"]["id"]
move(router, path, 1, f"merge-built-{key_suffix}", "merged", {"merge_commit": "3" * 40})
built = move(router, path, 2, f"build-built-{key_suffix}", "built", build_evidence())[1]
return path, built
def test_full_release_chain_is_exact_immutable_and_live_only_after_convergence(setup):
router, _, _, base = setup
status, created, headers = create(setup)
assert (status, created["revision"], headers["ETag"]) == (201, 1, "1")
assert contracts.validate("release.schema.json", created["release"]) == []
release_id = created["release"]["id"]
path = base + "/" + release_id
steps = [
("merged", {"merge_commit": "2" * 40}),
("built", build_evidence()),
("verified", {}),
("deployed", {"flux_revision": "main@sha1:" + "3" * 40}),
("converged", {"pod_digest": DIGEST}),
("live_verified", {"health_check": {"url": "/healthz", "status": "pass", "at": "2026-08-24T12:00:00Z"}}),
]
view = created
for revision, (target, evidence) in enumerate(steps, 1):
status, view, headers = move(router, path, revision, f"transition-{target}", target, evidence)
assert status == 200 and view["release"]["state"] == target
assert view["revision"] == revision + 1 and headers["ETag"] == str(revision + 1)
assert contracts.validate("release.schema.json", view["release"]) == []
assert view["release"]["evidence"]["image_digest"] == view["release"]["evidence"]["harbor_digest"] == view["release"]["evidence"]["pod_digest"]
rows = tenant(router).read(releases.FAMILY, release_id)
assert len(rows) == 7 and [row["sequence"] for row in rows] == list(range(1, 8))
assert all(contracts.validate("release-ledger.schema.json", row, SCHEMAS) == [] for row in rows)
assert all(row["previous_hash"] == (releases.ZERO_HASH if index == 0 else rows[index - 1]["entry_hash"]) for index, row in enumerate(rows))
assert call(router, "GET", path)[1] == view
listed = call(router, "GET", base)[1]
assert listed["items"] == [view]
def test_create_replay_uniqueness_and_scope(setup):
router, project, _, base = setup
status, created, _ = create(setup)
status, replay, headers = create(setup)
assert status == 200 and replay == created and headers["HUX-Replayed"] == "true"
assert create(setup, "release-key-001", "2" * 40)[0] == 409
assert create(setup, "release-key-002")[0] == 409
assert create(setup, "release-key-003", "2" * 40)[0] == 201
assert call(router, "GET", base + "/" + created["release"]["id"], headers=OTHER)[0] == 404
wrong = base.replace(project["id"], "prj_missing0000")
assert call(router, "GET", wrong + "/" + created["release"]["id"])[0] == 404
assert call(router, "GET", base + "/rel_missing0000")[0] == 404
assert call(router, "GET", base + "/BAD")[0] == 404
@pytest.mark.parametrize("changes", [
{"workload": "unknown"}, {"commit": "short"}, {"feature_flags": "hux.foundation"},
{"feature_flags": ["bad flag"]}, {"evidence": {}},
{"evidence": {"review_url": "http://git.bstein.dev/pr/1"}},
{"evidence": {"review_url": "https://user:pass@git.bstein.dev/pr/1"}},
{"evidence": {"review_url": "https://git.bstein.dev/pr/1?token=secret"}},
])
def test_create_validation_is_fail_closed(setup, changes):
commit = changes.get("commit", "4" * 40)
rest = {key: value for key, value in changes.items() if key != "commit"}
status, error, _ = create(setup, f"bad-release-{abs(hash(repr(changes)))}", commit, **rest)
assert status == 400 and error["code"] == "invalid"
def test_create_requires_headers_body_and_capacity(setup, monkeypatch):
router, _, _, base = setup
body = {"workload": "hermes-webui", "commit": "1" * 40, "evidence": {"review_url": "https://git.bstein.dev/pr/1"}}
assert call(router, "POST", base, body, {"Idempotency-Key": "no-if-match"})[0] == 400
assert call(router, "POST", base, body, {"If-Match": "0"})[0] == 400
assert router.dispatch("POST", base, HEADERS, b"[]").status == 400
assert call(router, "POST", base, {**body, "extra": True}, {"If-Match": "0", "Idempotency-Key": "extra-field-key"})[0] == 400
monkeypatch.setattr(releases, "MAX_RELEASES", 0)
assert call(router, "POST", base, body, {"If-Match": "0", "Idempotency-Key": "over-release-limit"})[0] == 413
def test_transition_requires_exact_order_revision_and_evidence(setup):
router, _, _, base = setup
created = create(setup)[1]
path = base + "/" + created["release"]["id"]
assert move(router, path, 1, "skip-transition", "built", build_evidence())[0] == 409
assert move(router, path, 1, "rollback-too-soon", "rolled_back", {"rollback_target": OTHER_DIGEST})[0] == 409
assert move(router, path, 2, "stale-transition", "merged", {"merge_commit": "2" * 40})[0] == 409
for index, evidence in enumerate(({}, {"merge_commit": "short"}, {"merge_commit": "2" * 40, "extra": "x"})):
assert move(router, path, 1, f"bad-merge-{index}", "merged", evidence)[0] == 400
status, merged, _ = move(router, path, 1, "good-merge-key", "merged", {"merge_commit": "2" * 40})
assert status == 200 and merged["revision"] == 2
status, replay, headers = move(router, path, 1, "good-merge-key", "merged", {"merge_commit": "2" * 40})
assert status == 200 and replay == merged and headers["HUX-Replayed"] == "true"
assert move(router, path, 2, "good-merge-key", "built", build_evidence())[0] == 409
assert call(router, "POST", path + "/transitions", {"to": "built", "evidence": build_evidence()}, {"Idempotency-Key": "missing-transition-if"})[0] == 400
@pytest.mark.parametrize("evidence", [
{**build_evidence(), "harbor_digest": OTHER_DIGEST},
{**build_evidence(), "image_digest": OTHER_DIGEST},
{**build_evidence(), "image_ref": "registry.bstein.dev/bstein/hermes:tag@" + OTHER_DIGEST},
{**build_evidence(), "ci_build_url": "http://jenkins.bstein.dev/job/1"},
{**build_evidence(), "ci_build_url": "https://jenkins.bstein.dev/job/1?token=x"},
{"ci_build_url": "https://jenkins.bstein.dev/job/1", "image_ref": "bad", "image_digest": DIGEST, "harbor_digest": DIGEST},
])
def test_build_requires_matching_image_and_harbor_evidence(setup, evidence):
router, _, _, base = setup
suffix = hashlib_key(evidence)
created = create(setup, f"create-{suffix}", (suffix * 4)[:40])[1]
path = base + "/" + created["release"]["id"]
assert move(router, path, 1, f"merge-{suffix}", "merged", {"merge_commit": "5" * 40})[0] == 200
assert move(router, path, 2, f"build-{suffix}", "built", evidence)[0] == 400
def hashlib_key(value):
import hashlib
return hashlib.sha256(repr(value).encode()).hexdigest()[:12]
def test_deploy_convergence_health_and_rollback_checks(setup):
router, _, _, _ = setup
path, built = through_built(setup)
assert built["release"]["state"] == "built"
assert move(router, path, 3, "deploy-before-verify", "deployed", {"flux_revision": "main@sha1:" + "3" * 40})[0] == 409
assert move(router, path, 3, "verify-with-data", "verified", {"harbor_digest": DIGEST})[0] == 400
assert move(router, path, 3, "verify-ok", "verified", {})[0] == 200
assert move(router, path, 4, "deploy-bad", "deployed", {"flux_revision": "feature@sha1:" + "3" * 40})[0] == 400
assert move(router, path, 4, "deploy-ok", "deployed", {"flux_revision": "main@sha1:" + "3" * 40})[0] == 200
assert move(router, path, 5, "converge-bad", "converged", {"pod_digest": OTHER_DIGEST})[0] == 400
assert move(router, path, 5, "converge-ok", "converged", {"pod_digest": DIGEST})[0] == 200
bad_health = [
{"health_check": {"url": "/healthz", "status": "fail", "at": "2026-08-24T12:00:00Z"}},
{"health_check": {"url": "http://public.example/health", "status": "pass", "at": "2026-08-24T12:00:00Z"}},
{"health_check": {"url": "/healthz?token=x", "status": "pass", "at": "2026-08-24T12:00:00Z"}},
]
for index, evidence in enumerate(bad_health):
assert move(router, path, 6, f"health-bad-{index}", "live_verified", evidence)[0] == 400
assert move(router, path, 6, "health-ok", "live_verified", {"health_check": {"url": "https://chat.bstein.dev/healthz", "status": "pass", "at": "2026-08-24T12:00:00Z"}})[0] == 200
assert move(router, path, 7, "after-live", "live_verified", {})[0] == 409
def test_rollback_is_terminal_and_digest_bound(setup):
router, _, _, _ = setup
path, _ = through_built(setup, "rollback", "6" * 40)
assert move(router, path, 3, "rollback-bad-digest", "rolled_back", {"rollback_target": "bad"})[0] == 400
status, rolled, _ = move(router, path, 3, "rollback-ok", "rolled_back", {"rollback_target": OTHER_DIGEST})
assert status == 200 and rolled["release"]["state"] == "rolled_back"
assert move(router, path, 4, "rollback-terminal", "verified", {})[0] == 409
def test_tamper_breaks_reads_and_idempotency_fails_closed(setup):
router, _, _, base = setup
created = create(setup)[1]
release_id = created["release"]["id"]
ledger = tenant(router).root / releases.FAMILY / f"{release_id}.jsonl"
rows = ledger.read_text().splitlines()
row = json.loads(rows[0])
row["snapshot"]["state"] = "live_verified"
ledger.write_text(json.dumps(row) + "\n")
status, error, _ = call(router, "GET", base + "/" + release_id)
assert status == 400 and "integrity" in error["message"]
status, error, _ = create(setup)
assert status == 400 and "integrity" in error["message"]
def test_inconsistent_idempotency_and_cross_conversation_views_fail_closed(setup):
router, project, _, base = setup
created = create(setup)[1]
idem = tenant(router).read(releases.FAMILY, "idempotency")
idem[0]["sequence"] = 99
tenant(router).rewrite(releases.FAMILY, "idempotency", idem)
status, error, _ = create(setup)
assert status == 400 and "idempotency ledger" in error["message"]
_, conversation2, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C2", "project_id": project["id"]})
base2 = f"/hux/v1/projects/{project['id']}/conversations/{conversation2['id']}/releases"
body = {"workload": "hermes-agent", "commit": "9" * 40, "evidence": {"review_url": "https://git.bstein.dev/pr/99"}}
second = call(router, "POST", base2, body, {"If-Match": "0", "Idempotency-Key": "second-scope-key"})[1]
first_id = created["release"]["id"]
second_id = second["release"]["id"]
assert call(router, "GET", base2 + "/" + first_id)[0] == 404
assert call(router, "GET", base + "/" + second_id)[0] == 404
assert move(router, base2 + "/" + first_id, 1, "cross-transition", "merged", {"merge_commit": "8" * 40})[0] == 404
listed = call(router, "GET", base)[1]
assert [item["release"]["id"] for item in listed["items"]] == [first_id]
def test_internal_helpers_are_bounded_and_worker_actor_has_no_subject():
worker = SimpleNamespace(identity=identity.Identity("slot-3", "usr_0123456789abcdef", "worker", "worker"))
assert releases._actor(worker) == {"type": "system", "id": "release-worker"}
with pytest.raises(Invalid, match="missing or too long"):
releases._safe_url(None)
with pytest.raises(Invalid, match="secret"):
releases._safe_url("https://git.bstein.dev/ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
with pytest.raises(Invalid):
releases._safe_url("https://git.bstein.dev/pr/1#fragment")
with pytest.raises(Invalid, match="ledger entry"):
releases._validate_entry({"schema": "broken"})
with pytest.raises(Invalid, match="does not match Harbor"):
releases._transition_evidence({"evidence": {"image_digest": DIGEST, "harbor_digest": OTHER_DIGEST}}, "verified", {})
for path in (FOUNDATION / "hux" / "releases.py", ROOT / "services" / "hermes" / "contracts" / "hux" / "release-ledger.schema.json", Path(__file__)):
assert len(path.read_text().splitlines()) <= 500

View File

@ -0,0 +1,183 @@
"""HUX-10 same-process retention scheduler lifecycle and failure bounds."""
from __future__ import annotations
import sys
import threading
from datetime import datetime, timezone
from http.client import HTTPConnection
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import identity, privacy
from hux.errors import Unauthorized
from hux.http import serve
from hux.retention_scheduler import RetentionScheduler, subject_binding
from hux.server import build_router
from hux.store import TenantStore
SUBJECT = "usr_0123456789abcdef"
NOW = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc)
FLAGS = "hux.foundation,hux.privacy"
def binding(tmp_path: Path, subject: str = SUBJECT, mode: int = 0o440) -> Path:
path = tmp_path / "subject-binding"
path.write_text(subject + "\n")
path.chmod(mode)
return path
def router_for(tmp_path: Path, path: Path | None, flags: str = FLAGS):
environ = {"HUX_FLAGS": flags, "HUX_TENANT_SLOT": "slot-3"}
if path is not None:
environ["HUX_SUBJECT_BINDING_FILE"] = str(path)
return build_router(tmp_path / "data", environ)
def test_scheduler_runs_stale_startup_then_stops_from_daily_wait(tmp_path):
router = router_for(tmp_path, binding(tmp_path))
calls = []
waits = []
def run(store, now, who):
calls.append((store.identity, now, who))
return privacy.run_retention(store, now, who)
scheduler = RetentionScheduler(
router, clock=lambda: NOW, runner=run, waiter=lambda delay: waits.append(delay) or True, daily_seconds=100
)
scheduler.run()
assert len(calls) == 1 and calls[0][0] == calls[0][2]
assert calls[0][0] == identity.Identity("slot-3", SUBJECT, "worker", "worker")
assert waits == [100 + scheduler._jitter(calls[0][0])]
assert scheduler.health() == {
"enabled": True,
"status": "stopped",
"last_run": "2026-08-24T12:00:00Z",
"last_success": "2026-08-24T12:00:00Z",
"errors": 0,
}
def test_fresh_startup_skips_work_but_next_daily_cycle_runs(tmp_path):
path = binding(tmp_path)
router = router_for(tmp_path, path)
who = subject_binding(router.environ)
privacy.run_retention(TenantStore(router.data_root, who), NOW, who)
calls = []
waits = []
def wait(delay):
waits.append(delay)
return len(waits) == 2
scheduler = RetentionScheduler(router, clock=lambda: NOW, runner=lambda *args: calls.append(args) or {}, waiter=wait, daily_seconds=20)
scheduler.run()
assert len(calls) == 1 and len(waits) == 2
assert all(delay == 20 + scheduler._jitter(who) for delay in waits)
def test_binding_and_job_errors_back_off_and_recover_without_api_failure(tmp_path):
path = tmp_path / "subject-binding"
router = router_for(tmp_path, path)
waits = []
attempts = []
def wait(delay):
waits.append(delay)
if len(waits) == 1:
binding(tmp_path)
return len(waits) == 3
def run(*args):
attempts.append(args)
if len(attempts) == 1:
raise RuntimeError("sensitive provider-looking detail")
return {}
scheduler = RetentionScheduler(router, clock=lambda: NOW, runner=run, waiter=wait, daily_seconds=30)
scheduler.run()
assert len(attempts) == 2
assert waits[:2] == [5.0, 10.0]
assert waits[2] == 30 + scheduler._jitter(subject_binding(router.environ))
health = scheduler.health()
assert health["status"] == "stopped" and health["errors"] == 0
assert "sensitive" not in str(health)
def test_clock_error_is_isolated_and_retried(tmp_path):
router = router_for(tmp_path, binding(tmp_path))
clocks = [RuntimeError("clock detail"), NOW]
waits = []
def clock():
value = clocks.pop(0)
if isinstance(value, Exception):
raise value
return value
scheduler = RetentionScheduler(router, clock=clock, runner=lambda *args: {}, waiter=lambda delay: waits.append(delay) or len(waits) > 1)
scheduler.run()
assert waits[0] == 5.0 and len(waits) == 2
assert scheduler.health()["last_success"] == "2026-08-24T12:00:00Z"
def test_server_close_wakes_and_joins_scheduler_and_health_exposes_state(tmp_path):
router = router_for(tmp_path, None)
server = serve(router, "127.0.0.1", 0)
serving = threading.Thread(target=server.serve_forever, daemon=True)
serving.start()
conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=5)
conn.request("GET", "/healthz")
body = __import__("json").loads(conn.getresponse().read())
assert body["status"] == "ok" and body["retention"]["enabled"] is True
assert body["retention"]["status"] in {"starting", "waiting_for_binding"}
server.shutdown()
server.server_close()
assert router.retention_scheduler.health()["status"] == "stopped"
assert not router.retention_scheduler._thread.is_alive()
def test_scheduler_is_off_unless_hux10_dependency_chain_is_enabled(tmp_path):
router = router_for(tmp_path, binding(tmp_path), "hux.foundation")
scheduler = RetentionScheduler(router)
scheduler.start()
assert scheduler.health()["status"] == "disabled" and scheduler._thread is None
scheduler.run()
def test_binding_reader_and_jitter_fail_closed_at_edge_cases(tmp_path, monkeypatch):
path = binding(tmp_path, mode=0o600)
router = router_for(tmp_path, path)
with pytest.raises(Exception, match="binding file unavailable"):
subject_binding(router.environ)
path.chmod(0o440)
router.environ["HUX_RETENTION_JITTER_SECONDS"] = "bad"
scheduler = RetentionScheduler(router)
who = subject_binding(router.environ)
assert 0 <= scheduler._jitter(who) <= 900
router.environ["HUX_RETENTION_JITTER_SECONDS"] = "0"
assert scheduler._jitter(who) == 0
monkeypatch.setattr(
"hux.retention_scheduler._read_subject_binding",
lambda unused: (_ for _ in ()).throw(Unauthorized("unsafe internal detail")),
)
with pytest.raises(Exception, match="binding file unavailable") as denied:
subject_binding(router.environ)
assert "unsafe internal" not in str(denied.value)
def test_scheduler_stop_is_idempotent_and_backoff_is_bounded(tmp_path):
scheduler = RetentionScheduler(router_for(tmp_path, binding(tmp_path)))
scheduler._thread = threading.current_thread()
scheduler.stop()
scheduler.stop()
assert scheduler.health()["status"] == "stopped"
assert scheduler._backoff(1000) == 640.0

View File

@ -0,0 +1,198 @@
"""Focused HUX-09 server-authoritative suggestion tests."""
from __future__ import annotations
import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import contracts, identity, privacy, store, suggestions
from hux.errors import Invalid
from hux.server import build_router
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
START = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc)
def call(router, method, path, body=None, headers=None):
raw = b"" if body is None else json.dumps(body).encode()
response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, raw)
return response.status, response.body, response.headers
def tenant(router):
return store.TenantStore(router.data_root, identity.resolve(HEADERS, {"HUX_ROUTER_KEY": "rk"}))
@pytest.fixture
def setup(tmp_path, monkeypatch):
moment = {"value": START}
monkeypatch.setattr(suggestions, "clock", lambda: moment["value"])
monkeypatch.setattr(suggestions, "now_iso", lambda: moment["value"].strftime("%Y-%m-%dT%H:%M:%SZ"))
router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
_, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "P"})
_, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
base = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}/suggestions"
return router, project, conversation, base, moment
def evaluate(setup, context="first_session", key="suggest-key-001", **extra):
router, _, _, base, _ = setup
return call(router, "POST", base + "/evaluate", {"context": context, **extra}, {"Idempotency-Key": key})
def test_every_context_returns_a_canonical_server_definition(setup):
for index, context in enumerate(suggestions.CONTEXTS):
status, result, headers = evaluate(setup, context, f"context-key-{index:02}")
assert status == 200 and result["stored"] is True and headers["Cache-Control"] == "no-store"
suggestion = result["suggestion"]
assert suggestion["trigger"] == {"surface": "chat", "context": context}
assert suggestion["suppression"] == {"dismissable": True, "max_shows": 3, "cooldown_seconds": 86400, "never_again_supported": True}
assert contracts.validate("suggestion.schema.json", suggestion, pointer="/$defs/suggestion") == []
assert contracts.validate("suggestion.schema.json", result["state"], pointer="/$defs/state") == []
def test_idempotency_cooldown_and_max_show_are_authoritative(setup):
_, _, _, _, moment = setup
status, first, _ = evaluate(setup)
assert status == 200 and first["state"]["shows"] == 1 and first["revision"] == 1
status, replay, headers = evaluate(setup)
assert status == 200 and replay == first and headers["HUX-Replayed"] == "true"
assert evaluate(setup, context="idle")[0] == 409
status, cooled, _ = evaluate(setup, key="suggest-key-002")
assert status == 200 and cooled == {"suggestion": None, "reason": "cooldown", "stored": False}
moment["value"] += timedelta(days=1, seconds=1)
second = evaluate(setup, key="suggest-key-003")[1]
assert second["state"]["shows"] == 2 and second["revision"] == 2
moment["value"] += timedelta(days=1, seconds=1)
third = evaluate(setup, key="suggest-key-004")[1]
assert third["state"]["shows"] == 3
moment["value"] += timedelta(days=1, seconds=1)
exhausted = evaluate(setup, key="suggest-key-005")[1]
assert exhausted == {"suggestion": None, "reason": "max_shows", "stored": False}
def test_decisions_require_click_if_match_and_idempotency(setup):
_, _, _, base, moment = setup
shown = evaluate(setup)[1]
suggestion_id = shown["suggestion"]["id"]
path = base + f"/{suggestion_id}/decisions"
headers = {"If-Match": "1", "Idempotency-Key": "decision-key-001"}
status, result, response_headers = call(setup[0], "POST", path, {"decision": "dismissed", "clicked": True}, headers)
assert status == 200 and result["decision"] == "dismissed" and result["revision"] == 2
assert response_headers["ETag"] == "2" and "dismissed_at" in result["state"]
status, replay, replay_headers = call(setup[0], "POST", path, {"decision": "dismissed", "clicked": True}, headers)
assert status == 200 and replay == result and replay_headers["HUX-Replayed"] == "true"
assert call(setup[0], "POST", path, {"decision": "acted", "clicked": True}, headers)[0] == 409
moment["value"] += timedelta(days=1, seconds=1)
shown_again = evaluate(setup, key="suggest-key-after-dismiss")[1]
assert shown_again["suggestion"]["id"] == suggestion_id and shown_again["state"]["shows"] == 2
invalid = [
({"decision": "acted", "clicked": False}, {"If-Match": "3", "Idempotency-Key": "decision-key-002"}),
({"decision": "silent", "clicked": True}, {"If-Match": "3", "Idempotency-Key": "decision-key-003"}),
({"decision": "acted", "clicked": True}, {"Idempotency-Key": "decision-key-004"}),
({"decision": "acted", "clicked": True}, {"If-Match": "3"}),
]
for body, bad_headers in invalid:
assert call(setup[0], "POST", path, body, bad_headers)[0] == 400
def test_never_again_is_explicit_and_permanent(setup):
_, _, _, base, moment = setup
shown = evaluate(setup)[1]
path = base + f"/{shown['suggestion']['id']}/decisions"
status, result, _ = call(setup[0], "POST", path, {"decision": "never_again", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "never-key-0001"})
assert status == 200 and result["state"]["never_again"] is True
moment["value"] += timedelta(days=365)
blocked = evaluate(setup, key="suggest-key-never")[1]
assert blocked == {"suggestion": None, "reason": "never_again", "stored": False}
@pytest.mark.parametrize(("gate", "reason"), [
("no_store", "client_no_store"), ("sensitive", "client_no_store"), ("restricted", "client_no_store"),
])
def test_client_privacy_gates_do_not_write_state(setup, gate, reason):
body = {"no_store": True} if gate == "no_store" else {"sensitivity": gate}
status, result, _ = evaluate(setup, key=f"privacy-{gate}-1", **body)
assert status == 200 and result == {"suggestion": None, "reason": reason, "stored": False}
assert tenant(setup[0]).count(suggestions.FAMILY) == 0
assert tenant(setup[0]).read(suggestions.FAMILY, "idempotency") == []
def test_server_privacy_state_overrides_client_claims(setup):
router, _, conversation, base, _ = setup
privacy.mark_topic(tenant(router), conversation["id"], "health")
result = evaluate(setup, key="privacy-topic-01", sensitivity="public")[1]
assert result["reason"] == "sensitive_topic" and tenant(router).count(suggestions.FAMILY) == 0
privacy.set_flag(tenant(router), conversation["id"], "memory_disabled", True)
assert evaluate(setup, key="privacy-disabled", sensitivity="public")[1]["reason"] == "conversation_no_store"
privacy.set_flag(tenant(router), conversation["id"], "forgotten", True)
assert evaluate(setup, key="privacy-forgotten", sensitivity="public")[1]["reason"] == "conversation_no_store"
privacy.set_flag(tenant(router), conversation["id"], "forgotten", False)
privacy.set_flag(tenant(router), conversation["id"], "memory_disabled", False)
doc = tenant(router).get("privacy", privacy.TOPICS_DOC)
doc["items"][conversation["id"]]["topics"] = []
tenant(router).put("privacy", doc)
current = tenant(router).get("conversations", conversation["id"])
tenant(router).put("conversations", {**current, "mode": "private"}, current["revision"])
result = call(router, "POST", base + "/evaluate", {"context": "idle", "sensitivity": "public"}, {"Idempotency-Key": "privacy-private1"})[1]
assert result["reason"] == "private_mode"
def test_list_state_scope_and_limits(setup, monkeypatch):
router, project, _, base, _ = setup
shown = evaluate(setup)[1]
status, states, headers = call(router, "GET", base + "/states")
assert status == 200 and states["items"][0]["suggestion_id"] == shown["suggestion"]["id"]
assert states["items"][0]["revision"] == 1 and headers["Cache-Control"] == "no-store"
assert call(router, "GET", base + "/states", headers=OTHER)[0] == 404
wrong = base.replace(project["id"], "prj_missing0000")
assert call(router, "GET", wrong + "/states")[0] == 404
_, conversation2, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C2", "project_id": project["id"]})
base2 = f"/hux/v1/projects/{project['id']}/conversations/{conversation2['id']}/suggestions"
monkeypatch.setattr(suggestions, "MAX_STATES", 0)
result = call(router, "POST", base2 + "/evaluate", {"context": "idle"}, {"Idempotency-Key": "state-limit-key"})[1]
assert result == {"suggestion": None, "reason": "state_limit", "stored": False}
def test_bad_context_body_and_unknown_decision_fail_closed(setup):
router, _, _, base, _ = setup
for body in ([], {"context": "nope"}, {"context": "idle", "priority": 100}):
raw = json.dumps(body).encode()
assert router.dispatch("POST", base + "/evaluate", {**HEADERS, "Idempotency-Key": "bad-context-key"}, raw).status == 400
assert call(router, "POST", base + "/evaluate", {"context": "idle"})[0] == 400
missing = base + "/sug_missing0000/decisions"
assert call(router, "POST", missing, {"decision": "acted", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "missing-decision"})[0] == 404
def test_defensive_validation_stale_act_and_private_inspection(setup):
router, _, conversation, base, _ = setup
with pytest.raises(Invalid, match="suggestion failed"):
suggestions._suggestion("idle", "invalid-surface")
with pytest.raises(Invalid, match="state failed"):
suggestions._public({"schema": "hux.suggestion_state.v1", "owner": "raw-user", "suggestion_id": "sug_idle_workflow", "shows": 1, "never_again": False})
shown = evaluate(setup)[1]
path = base + f"/{shown['suggestion']['id']}/decisions"
assert call(router, "POST", path, {"decision": "acted", "clicked": True}, {"If-Match": "9", "Idempotency-Key": "stale-decision1"})[0] == 409
status, acted, _ = call(router, "POST", path, {"decision": "acted", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "acted-decision1"})
assert status == 200 and "acted_at" in acted["state"] and "dismissed_at" not in acted["state"]
current = tenant(router).get("conversations", conversation["id"])
tenant(router).put("conversations", {**current, "mode": "private"}, current["revision"])
assert call(router, "POST", path, {"decision": "dismissed", "clicked": True}, {"If-Match": "2", "Idempotency-Key": "private-decision"})[0] == 404
status, states, _ = call(router, "GET", base + "/states")
assert status == 200 and states == {"items": [], "next": None}
def test_source_and_test_files_stay_bounded():
assert len((FOUNDATION / "hux" / "suggestions.py").read_text().splitlines()) <= 500
assert len(Path(__file__).read_text().splitlines()) <= 500