API follow-on: bounded worker admission and scoped usage controls

c3r/adapters/base.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Mapping, Protocol
+from typing import Protocol
 
 from ..state_schema import ActionCandidate
 

c3r/adapters/providers.py

@@ -17,6 +17,7 @@
 from ..deliberative.envelope import DeliberativeResult
 from ..http_transport import NoRedirectHandler
 from ..state_schema import ActionFamily
+from ..telemetry.invocations import transport_attempt
 
 MAX_RESPONSE_BYTES = 65_536
 MAX_ARRAY_ITEMS = 32
@@ -396,6 +397,7 @@ def generate(self, text: str, max_output_tokens: int) -> tuple[str, str, dict[st
             payload["provider"] = {"zdr": True, "data_collection": "deny",
                                    "require_parameters": True, "allow_fallbacks": False}
         try:
+            transport_attempt("system_two")
             response = self._transport(self.config.base_url.rstrip("/") + "/chat/completions",
                                        headers, payload)
             size = len(json.dumps(response.body, allow_nan=False).encode())
@@ -479,6 +481,7 @@ def interrupt_opening() -> None:
         try:
             if cancelled is not None and cancelled.is_set():
                 raise RuntimeError("generation cancelled")
+            transport_attempt("system_two")
             connection.connect()
             backend_socket = connection.sock
             if backend_socket is None:
@@ -508,6 +511,7 @@ def interrupt_opening() -> None:
     def deliberate(self, request: DeliberationRequest) -> ProviderExecutionResult:
         state = json.dumps(dict(request.state), sort_keys=True, separators=(",", ":"))
         url, headers, payload = self._codec.build(self.config, state)
+        transport_attempt("system_two")
         response = self._transport(url, headers, payload)
         if len(json.dumps(response.body, separators=(",", ":")).encode()) > MAX_RESPONSE_BYTES:
             raise ValueError("provider response exceeds byte limit")

c3r/api_access.py

@@ -45,7 +45,8 @@ def _identifier(value: str) -> None:
 
 
 class AccessStore:
-    def __init__(self, path: Path, *, clock: Callable[[], float] = time.time) -> None:
+    @staticmethod
+    def validate_path(path: Path) -> None:
         if (not path.is_absolute() or not path.parent.is_dir()
                 or any(p.is_symlink() for p in (path, *path.parents))):
             raise ValueError("non-linked absolute access database path and existing directory required")
@@ -56,6 +57,9 @@ def __init__(self, path: Path, *, clock: Callable[[], float] = time.time) -> Non
                 unsafe_mode = bool(info.st_mode & (0o077 if ancestor == path.parent else 0o022))
                 if info.st_uid not in {0, os.geteuid()} or (unsafe_mode and not sticky_root):
                     raise ValueError("private operator-owned access database directory required")
+
+    def __init__(self, path: Path, *, clock: Callable[[], float] = time.time) -> None:
+        self.validate_path(path)
         if not path.exists():
             descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
             os.close(descriptor)
@@ -148,8 +152,11 @@ def authenticate(self, plaintext: str) -> Principal | None:
 
     def revoke_key(self, tenant: str, project: str, key_id: str) -> None:
         with self.connect() as connection:
-            connection.execute("UPDATE api_keys SET revoked=1 WHERE tenant=? AND project=? AND id=?",
-                               (tenant, project, key_id))
+            changed = connection.execute(
+                "UPDATE api_keys SET revoked=1 WHERE tenant=? AND project=? AND id=?",
+                (tenant, project, key_id)).rowcount
+            if changed != 1:
+                raise ValueError("key not found in selected project")
             connection.execute("INSERT INTO audit_events VALUES (NULL,?,?,?,'key_revoked',NULL,?)",
                                (tenant, project, key_id, self.clock()))
 
@@ -171,14 +178,16 @@ def record_usage(self, principal: Principal, request_id: str, *, route: str,
                      model: str | None, status: int, latency_ms: float,
                      input_tokens: int | None, output_tokens: int | None,
                      system_one_invocations: int | None,
-                     system_two_invocations: int | None) -> None:
+                     system_two_invocations: int | None,
+                     invocation_basis: str | None = None) -> None:
         metadata = {"request_id": request_id, "tenant_id": principal.tenant_id,
                     "project_id": principal.project_id, "api_key_id": principal.key_id,
                     "route": route, "model": model, "status": status,
                     "latency_ms": latency_ms, "timestamp": self.clock(),
                     "input_tokens": input_tokens, "output_tokens": output_tokens,
                     "system_one_invocations": system_one_invocations,
                     "system_two_invocations": system_two_invocations,
+                    "invocation_basis": invocation_basis,
                     "gpu_allocation_ms": None, "allocated_cost_usd": None,
                     "cost_basis": "unmeasured"}
         with self.connect() as connection:
@@ -192,6 +201,30 @@ def project_usage(self, tenant: str, project: str) -> list[dict[str, object]]:
                                       (tenant, project)).fetchall()
             return [cast(dict[str, object], json.loads(row[0])) for row in rows]
 
+    def purge_metadata(self, tenant: str, project: str, *, retention_seconds: int,
+                       limit: int = 1000) -> dict[str, object]:
+        """Bounded logical deletion in this database only; not backup/physical erase proof."""
+        _identifier(tenant)
+        _identifier(project)
+        if (type(retention_seconds) is not int or not 60 <= retention_seconds <= 30 * 86400
+                or type(limit) is not int or not 1 <= limit <= 10000):
+            raise ValueError("bounded metadata retention and batch size required")
+        cutoff = self.clock() - retention_seconds
+        with self.connect() as connection:
+            connection.execute("PRAGMA secure_delete=ON")
+            connection.execute("BEGIN IMMEDIATE")
+            usage = connection.execute(
+                "DELETE FROM usage_records WHERE rowid IN (SELECT rowid FROM usage_records "
+                "WHERE tenant=? AND project=? AND json_extract(metadata,'$.timestamp')<? "
+                "ORDER BY rowid LIMIT ?)", (tenant, project, cutoff, limit)).rowcount
+            audit = connection.execute(
+                "DELETE FROM audit_events WHERE id IN (SELECT id FROM audit_events "
+                "WHERE tenant=? AND project=? AND at<? ORDER BY id LIMIT ?)",
+                (tenant, project, cutoff, limit)).rowcount
+        return {"usage_deleted": usage, "audit_deleted": audit, "cutoff_utc_seconds": cutoff,
+                "batch_limit_per_table": limit, "scope": "selected_project_live_database_only",
+                "backup_deletion_verified": False, "physical_erasure_verified": False}
+
     def admit(self, principal: Principal) -> bool:
         now = int(self.clock())
         with self.connect() as connection:

c3r/candidate_compiler.py

@@ -6,14 +6,14 @@
 from collections.abc import Iterable
 from dataclasses import replace
 
+from .progressive_widening import widen_when_margin_is_small
 from .state_schema import (
     ActionCandidate,
     ActionDefinition,
     ActionFamily,
     AuthorityPolicy,
     CandidateCompilation,
 )
-from .progressive_widening import widen_when_margin_is_small
 
 
 class CandidateCompiler:
@@ -160,9 +160,11 @@ def compile_hierarchical(
                     maximum_width=min(maximum_width, self._per_family_cap),
                     margin=widening_margin,
                 )
-                if len(selected_operations) > min(initial_width, len(operation_skeletons)):
-                    if family not in widened:
-                        widened.append(family)
+                if (
+                    len(selected_operations) > min(initial_width, len(operation_skeletons))
+                    and family not in widened
+                ):
+                    widened.append(family)
                 definitions_by_id = {item.id: item for item in subgroup_definitions}
                 for selected_skeleton in selected_operations:
                     definition = definitions_by_id[selected_skeleton.id]

c3r/commit_gateway.py

@@ -2,10 +2,10 @@
 
 from __future__ import annotations
 
+import time
 from collections.abc import Callable
 from dataclasses import dataclass
 from threading import Lock
-import time
 from typing import Protocol
 
 from .authority import action_fingerprint, attestation_matches

c3r/cost_twin.py

@@ -17,8 +17,8 @@ class AdaptiveCostTwin:
     priors: dict[str, CostForecast]
     residual_alpha: float = 0.2
     stale_error_threshold: float = 0.5
-    _residuals: dict[str, float] = field(default_factory=dict, init=False)
-    _stale: set[str] = field(default_factory=set, init=False)
+    _residuals: dict[str, float] = field(default_factory=dict[str, float], init=False)
+    _stale: set[str] = field(default_factory=set[str], init=False)
 
     def forecast(self, action_id: str) -> CostForecast:
         prior = self.priors[action_id]
@@ -37,4 +37,3 @@ def observe(self, action_id: str, observed_cost: float) -> None:
         relative_error = abs(error) / max(forecast.expected, 1e-9)
         if relative_error > self.stale_error_threshold:
             self._stale.add(action_id)
-

c3r/credential_recovery.py

@@ -0,0 +1,100 @@
+"""Read-only credential backup and hash-bound offline recovery; no payload copies."""
+import hashlib
+import hmac
+import os
+import re
+import sqlite3
+from collections.abc import Generator
+from contextlib import contextmanager
+from pathlib import Path
+
+from .api_access import AccessStore
+
+MAX_SNAPSHOT_BYTES = 64 * 1024 * 1024
+CREDENTIAL_TABLES = (
+    ("organizations", "id"),
+    ("projects", "tenant,id,rpm,key_rps"),
+    ("api_keys", "id,salt,digest,tenant,project,scopes,created_at,expires_at,last_used_at,revoked"),
+)
+
+
+@contextmanager
+def credential_source(path: Path, expected_sha256: str | None) -> Generator[sqlite3.Connection]:
+    AccessStore.validate_path(path)
+    if path.resolve() != path or not path.is_file() or path.stat().st_nlink != 1:
+        raise ValueError("existing unlinked credential source required")
+    if os.name == "posix" and (path.stat().st_uid != os.geteuid() or path.stat().st_mode & 0o077):
+        raise ValueError("private credential source ownership/mode required")
+    if expected_sha256 is None:
+        connection = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=2)
+    else:
+        if re.fullmatch(r"[a-f0-9]{64}", expected_sha256) is None:
+            raise ValueError("approved backup hash required")
+        for suffix in ("-wal", "-shm", "-journal"):
+            sidecar = path.with_name(path.name + suffix)
+            if sidecar.exists() or sidecar.is_symlink():
+                raise ValueError("unbound snapshot sidecar")
+        with path.open("rb") as artifact:
+            raw = artifact.read(MAX_SNAPSHOT_BYTES + 1)
+        if len(raw) > MAX_SNAPSHOT_BYTES:
+            raise ValueError("credential snapshot exceeds recovery bound")
+        if not hmac.compare_digest(hashlib.sha256(raw).hexdigest(), expected_sha256):
+            raise ValueError("credential snapshot hash mismatch")
+        # SQLite receives only these hash-verified bytes, never the pathname or
+        # journal state. Concurrent source mutation cannot change the restored data.
+        connection = sqlite3.connect(":memory:", timeout=2)
+        try:
+            connection.deserialize(raw)
+        except sqlite3.Error:
+            connection.close()
+            raise
+    try:
+        connection.execute("PRAGMA query_only=ON")
+        connection.execute("BEGIN")
+        if connection.execute("PRAGMA quick_check").fetchone() != ("ok",):
+            raise ValueError("credential source integrity check failed")
+        for table, columns in CREDENTIAL_TABLES:
+            connection.execute(f"SELECT {columns} FROM {table} LIMIT 0")
+        if expected_sha256 is not None and connection.execute(
+                "SELECT version FROM credential_snapshot").fetchall() != [(1,)]:
+            raise ValueError("versioned credential snapshot required")
+        yield connection
+    finally:
+        connection.close()
+
+
+def copy_credentials(source_path: Path, destination: Path, *,
+                     expected_sha256: str | None = None) -> dict[str, object]:
+    """Copy credentials into a new file; restore always revokes keys for secure reissue."""
+    restore = expected_sha256 is not None
+    AccessStore.validate_path(destination)
+    if destination.resolve() != destination:
+        raise ValueError("non-redirected destination required")
+    with credential_source(source_path, expected_sha256) as source:
+        descriptor = os.open(destination, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        os.close(descriptor)
+        target = AccessStore(destination)
+        counts: dict[str, int] = {}
+        with target.connect() as output:
+            for table, columns in CREDENTIAL_TABLES:
+                rows = source.execute(f"SELECT {columns} FROM {table}")
+                placeholders = ",".join("?" for _ in columns.split(","))
+                output.executemany(f"INSERT INTO {table} ({columns}) VALUES ({placeholders})", rows)
+                counts[table] = output.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
+            if restore:
+                output.execute("UPDATE api_keys SET revoked=1")
+            else:
+                output.execute("CREATE TABLE credential_snapshot (version INTEGER NOT NULL)")
+                output.execute("INSERT INTO credential_snapshot VALUES (1)")
+            if output.execute("PRAGMA foreign_key_check").fetchone() is not None:
+                raise ValueError("credential relationships invalid")
+    with destination.open("r+b") as artifact:
+        os.fsync(artifact.fileno())
+        if os.fstat(artifact.fileno()).st_size > MAX_SNAPSHOT_BYTES:
+            raise ValueError("credential snapshot exceeds recovery bound")
+        digest = hashlib.file_digest(artifact, "sha256").hexdigest()
+    return {"status": "restored_keys_revoked" if restore else "credential_snapshot_created",
+            "organizations_copied": counts["organizations"], "projects_copied": counts["projects"],
+            "keys_copied": counts["api_keys"], "sha256": digest,
+            "requires_key_rotation": restore, "scope": "credentials_and_project_limits_only",
+            "encryption_verified": False, "backup_deletion_verified": False}

c3r/decisionmix/dataset.py

@@ -2,13 +2,12 @@
 
 from __future__ import annotations
 
-from dataclasses import asdict, dataclass
 import hashlib
 import json
-from pathlib import Path
 import re
-from typing import Mapping
-
+from collections.abc import Mapping
+from dataclasses import asdict, dataclass
+from pathlib import Path
 
 SCHEMA_VERSION = "c3r.decisionmix.v1"
 SPLITS = ("train", "validation", "test")
@@ -57,7 +56,7 @@ def validate(self) -> None:
 
 
 def deterministic_split(record_id: str, *, seed: str) -> str:
-    digest = hashlib.sha256(f"{seed}\x00{record_id}".encode("utf-8")).digest()
+    digest = hashlib.sha256(f"{seed}\x00{record_id}".encode()).digest()
     bucket = int.from_bytes(digest[:8], "big") % 100
     if bucket < 80:
         return "train"

c3r/deliberative/__init__.py

@@ -1,9 +1,19 @@
 """Interfaces for open and frontier deliberative models."""
 
 from importlib import import_module
+from typing import TYPE_CHECKING
 
 from .envelope import DeliberativeEnvelope, DeliberativeResult
 
+if TYPE_CHECKING:
+    from .defaults import (
+        DEFAULT_DEEPSEEK_API_MODEL,
+        DEFAULT_LANGUAGE_MODEL,
+        DEFAULT_OPENROUTER_MODEL,
+        DefaultModelProfile,
+        default_provider_config,
+    )
+
 
 def __getattr__(name: str):
     if name in {

c3r/host_factory.py

@@ -4,12 +4,12 @@
 
 from collections.abc import Callable, Mapping
 from math import isfinite
+from typing import cast
 from uuid import uuid4
 
 from .runtime import RuntimeRequest
 from .state_schema import ActionDefinition, AuthorityPolicy, RawState, RiskClass, ValueEstimate
 
-
 EstimateSource = Callable[[RawState], Mapping[str, ValueEstimate]]
 
 
@@ -49,9 +49,10 @@ def build(self, payload: Mapping[str, object]) -> RuntimeRequest:
         goal = _text(payload.get("goal"), "goal")
         current_subgoal = _text(payload.get("current_subgoal"), "current_subgoal")
         questions = payload.get("open_questions", [])
-        if not isinstance(questions, list) or len(questions) > 64:
+        if not isinstance(questions, list) or len(cast(list[object], questions)) > 64:
             raise ValueError("open_questions must be a bounded array")
-        open_questions = tuple(_text(item, "open_question") for item in questions)
+        question_values = cast(list[object], questions)
+        open_questions = tuple(_text(item, "open_question") for item in question_values)
         raw = RawState(
             goal=goal,
             current_subgoal=current_subgoal,
@@ -63,9 +64,7 @@ def build(self, payload: Mapping[str, object]) -> RuntimeRequest:
             budget={"remaining_usd": self._remaining_usd},
             data_boundary=self._data_boundary,
         )
-        estimates = self._estimate_source(raw)
-        if not isinstance(estimates, Mapping):
-            raise ValueError("host estimate source returned an invalid mapping")
+        estimates = _host_estimates(self._estimate_source(raw))
         return RuntimeRequest(
             raw_state=raw,
             definitions=self._definitions,
@@ -79,3 +78,12 @@ def _text(value: object, name: str) -> str:
     if not isinstance(value, str) or not value.strip() or len(value) > 4096:
         raise ValueError(f"{name} must be non-empty bounded text")
     return value
+
+
+def _host_estimates(value: object) -> Mapping[str, ValueEstimate]:
+    # Keep the runtime shape check even for an incorrectly implemented host
+    # callback; this does not claim to validate individual estimate values.
+    estimates = cast(Mapping[str, ValueEstimate], value) if isinstance(value, Mapping) else None
+    if estimates is None:
+        raise ValueError("host estimate source returned an invalid mapping")
+    return estimates