diff --git a/c3r/adapters/base.py b/c3r/adapters/base.py index 65ebd0a..5963867 100644 --- a/c3r/adapters/base.py +++ b/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 diff --git a/c3r/adapters/providers.py b/c3r/adapters/providers.py index 1f00503..7ba9d8b 100644 --- a/c3r/adapters/providers.py +++ b/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") diff --git a/c3r/api_access.py b/c3r/api_access.py index 52efd06..3a312a5 100644 --- a/c3r/api_access.py +++ b/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,7 +178,8 @@ 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, @@ -179,6 +187,7 @@ def record_usage(self, principal: Principal, request_id: str, *, route: str, "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') bool: now = int(self.clock()) with self.connect() as connection: diff --git a/c3r/candidate_compiler.py b/c3r/candidate_compiler.py index a780034..fa514e7 100644 --- a/c3r/candidate_compiler.py +++ b/c3r/candidate_compiler.py @@ -6,6 +6,7 @@ from collections.abc import Iterable from dataclasses import replace +from .progressive_widening import widen_when_margin_is_small from .state_schema import ( ActionCandidate, ActionDefinition, @@ -13,7 +14,6 @@ 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] diff --git a/c3r/commit_gateway.py b/c3r/commit_gateway.py index 2f5d893..f4d3ca2 100644 --- a/c3r/commit_gateway.py +++ b/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 diff --git a/c3r/cost_twin.py b/c3r/cost_twin.py index 392bc32..9ab27c6 100644 --- a/c3r/cost_twin.py +++ b/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) - diff --git a/c3r/credential_recovery.py b/c3r/credential_recovery.py new file mode 100644 index 0000000..b363014 --- /dev/null +++ b/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} diff --git a/c3r/decisionmix/dataset.py b/c3r/decisionmix/dataset.py index bbf6fd5..fe9fdb8 100644 --- a/c3r/decisionmix/dataset.py +++ b/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" diff --git a/c3r/deliberative/__init__.py b/c3r/deliberative/__init__.py index d3d9a1c..dbac7f6 100644 --- a/c3r/deliberative/__init__.py +++ b/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 { diff --git a/c3r/host_factory.py b/c3r/host_factory.py index eccf63f..406a705 100644 --- a/c3r/host_factory.py +++ b/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 diff --git a/c3r/http_service.py b/c3r/http_service.py index 8f4f405..a08294f 100644 --- a/c3r/http_service.py +++ b/c3r/http_service.py @@ -23,6 +23,7 @@ from .responses import ResponseEventStream, ResponsesService from .runtime import RuntimeRequest, StandaloneController from .system_one.inference import SystemOneInference +from .telemetry.invocations import capture_invocations, with_invocations MAX_REQUEST_BYTES = 65_536 @@ -129,6 +130,7 @@ def log_message(self, format: str, *_args: object) -> None: return def _send(self, status: int, value: Mapping[str, object]) -> None: + value = with_invocations(value) body = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") @@ -227,6 +229,9 @@ def watch_disconnect() -> None: for name, data in events: if client_gone.is_set(): break + response = data.get("response") + if isinstance(response, dict): + data = {**data, "response": with_invocations(cast(Mapping[str, object], response))} frame = ("event: " + name + "\n" + "data: " + json.dumps(data, separators=(",", ":"), ensure_ascii=False) + "\n\n").encode("utf-8") @@ -288,6 +293,10 @@ def do_GET(self) -> None: self._send(404, {"error": "not_found"}) def do_POST(self) -> None: + with capture_invocations(): + self._post() + + def _post(self) -> None: if self.path not in { "/v1/decisions", "/v1/c3r/decide", "/v1/c3r/rank", "/v1/system-one", "/v1/c3r/execute", "/v1/responses", diff --git a/c3r/ingress_proxy.py b/c3r/ingress_proxy.py index fc84ba6..1e36476 100644 --- a/c3r/ingress_proxy.py +++ b/c3r/ingress_proxy.py @@ -41,6 +41,7 @@ def __init__( port: int = 8080, requests_per_minute: int = 60, max_in_flight: int = 16, + max_connections: int = 32, upstream_timeout_seconds: float = 5.0, access_store: AccessStore | None = None, ) -> None: @@ -54,6 +55,9 @@ def __init__( raise ValueError("client and upstream tokens must be different") if max_in_flight < 1 or upstream_timeout_seconds <= 0: raise ValueError("concurrency and timeout must be positive") + if type(max_connections) is not int or not 1 <= max_connections <= 4096: + raise ValueError("bounded accepted connection limit required") + self.connection_slots = BoundedSemaphore(max_connections) self.upstream_host = upstream_host self.upstream_port = upstream_port self.client_token = client_token @@ -78,6 +82,29 @@ def get_request(self) -> tuple[socket.socket, object]: connection.settimeout(5.0) return connection, address + def process_request(self, request: socket.socket | tuple[bytes, socket.socket], client_address: object) -> None: + # Refuse before spawning a handler or parsing untrusted headers/bodies. + # This is a transport refusal, not an authenticated API usage outcome. + if not self.connection_slots.acquire(blocking=False): + self.shutdown_request(request) + return + try: + super().process_request(request, client_address) + except BaseException: + self.connection_slots.release() + raise + + def process_request_thread(self, request: socket.socket | tuple[bytes, socket.socket], client_address: object) -> None: + try: + super().process_request_thread(request, client_address) + finally: + self.connection_slots.release() + + def handle_error(self, request: socket.socket | tuple[bytes, socket.socket], client_address: object) -> None: + # Socketserver's default traceback may contain caller-controlled exception + # diagnostics. Closed/malformed transports must never export that data. + return + class _IngressHandler(BaseHTTPRequestHandler): @property @@ -86,6 +113,7 @@ def gateway(self) -> C3RIngressServer: principal: Principal | None = None request_id: str = "" started: float = 0 + usage_recorded: bool = False def log_message(self, format: str, *args: object) -> None: return @@ -97,6 +125,12 @@ def send_error(self, code: int, message: str | None = None, explain: str | None "method_not_allowed" if code == 501 else "invalid_request") def _send_error(self, status: int, code: str) -> None: + try: + self._record_usage(status, {}) + except (OSError, sqlite3.Error): + # Store unavailability must still yield a sanitized refusal, not a + # second exception or an invented durable accounting record. + status, code = 503, "access_unavailable" value: dict[str, object] = {"error": code} if self.gateway.access_store is not None: category = {400: "invalid_request_error", 401: "authentication_error", @@ -120,6 +154,7 @@ def _begin_request(self) -> None: self.request_id = "req_" + uuid4().hex self.principal = None self.started = time.monotonic() + self.usage_recorded = False def _scope_allowed(self) -> bool: if self.gateway.access_store is None: @@ -136,7 +171,7 @@ def _admitted(self) -> bool: return self.principal is not None and self.gateway.access_store.admit(self.principal) def _record_usage(self, status: int, decoded: Mapping[str, object]) -> None: - if self.gateway.access_store is None or self.principal is None: + if self.gateway.access_store is None or self.principal is None or self.usage_recorded: return raw_usage = decoded.get("usage") raw_c3r = decoded.get("c3r") @@ -154,7 +189,10 @@ def count(value: object) -> int | None: input_tokens=count(usage.get("input_tokens")), output_tokens=count(usage.get("output_tokens")), system_one_invocations=count(c3r.get("system_one_invocations")), system_two_invocations=count(c3r.get("system_two_invocations")), + invocation_basis=("adapter_transport_attempts" + if c3r.get("invocation_basis") == "adapter_transport_attempts" else None), ) + self.usage_recorded = True def _authorized(self) -> bool: # Tenant and project identity come only from the authenticated key. @@ -246,7 +284,12 @@ def watch_disconnect() -> None: self._send_error(response.status, "upstream_rejected") return except (OSError, HTTPException, ValueError, sqlite3.Error): - if not disconnected.is_set(): + if disconnected.is_set(): + try: + self._record_usage(499, {}) + except (OSError, sqlite3.Error): + pass + else: self._send_error(503, "upstream_unavailable") return finally: diff --git a/c3r/key_management.py b/c3r/key_management.py index 2c7fd07..c1ff77a 100644 --- a/c3r/key_management.py +++ b/c3r/key_management.py @@ -10,13 +10,19 @@ from pathlib import Path from .api_access import AccessStore +from .credential_recovery import copy_credentials def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--database", type=Path, required=True) commands = parser.add_subparsers(dest="command", required=True) - for name in ("project", "issue", "keys", "revoke", "usage"): + for name in ("backup", "restore"): + command = commands.add_parser(name) + command.add_argument("--destination", type=Path, required=True) + if name == "restore": + command.add_argument("--expected-sha256", required=True) + for name in ("project", "issue", "keys", "revoke", "usage", "purge"): command = commands.add_parser(name) command.add_argument("--tenant", required=True) command.add_argument("--project", required=True) @@ -29,8 +35,17 @@ def main() -> int: command.add_argument("--ttl-seconds", type=int, default=86400) elif name == "revoke": command.add_argument("--key-id", required=True) + elif name == "purge": + command.add_argument("--retention-seconds", type=int, required=True) + command.add_argument("--limit", type=int, default=1000) args = parser.parse_args() try: + if args.command in {"backup", "restore"}: + result = copy_credentials(args.database, args.destination, + expected_sha256=args.expected_sha256 + if args.command == "restore" else None) + print(json.dumps(result, sort_keys=True)) + return 0 store = AccessStore(args.database) if args.command == "project": store.create_project(args.tenant, args.project, rpm=args.rpm, key_rps=args.key_rps) @@ -46,6 +61,9 @@ def main() -> int: result = {"status": "revoked"} elif args.command == "keys": result = store.list_keys(args.tenant, args.project) + elif args.command == "purge": + result = store.purge_metadata(args.tenant, args.project, + retention_seconds=args.retention_seconds, limit=args.limit) else: result = store.project_usage(args.tenant, args.project) print(json.dumps(result, sort_keys=True)) diff --git a/c3r/production_host.py b/c3r/production_host.py index 013c5a9..8eb56e8 100644 --- a/c3r/production_host.py +++ b/c3r/production_host.py @@ -48,9 +48,10 @@ def _get(self, url: str) -> Mapping[str, object]: if len(body) > 65536: raise ValueError("oversized provider readback") value = json.loads(body) - if not isinstance(value, dict): + mapping = cast(Mapping[str, object], value) if isinstance(value, dict) else None + if mapping is None: raise ValueError("invalid provider readback") - return cast(Mapping[str, object], value) + return mapping def _system_one(self) -> bool: try: diff --git a/c3r/retention_job.py b/c3r/retention_job.py index 2f87f59..37a2e8b 100644 --- a/c3r/retention_job.py +++ b/c3r/retention_job.py @@ -12,12 +12,11 @@ import re from collections.abc import Mapping from dataclasses import dataclass -from datetime import datetime, timedelta, timezone -from typing import Protocol +from datetime import UTC, datetime, timedelta +from typing import Protocol, cast from urllib.parse import quote, urlencode from urllib.request import Request, urlopen - DELETE_AFTER_DAYS = 28 _BUCKET = re.compile(r"colomboai-c3r-staging-traces-([0-9]{12})\Z") _METADATA_PROJECT_NUMBER = ( @@ -62,10 +61,10 @@ def _verify_runtime_project(bucket: str, project_number: str) -> None: def _created_at(raw: str) -> datetime: - parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + parsed = datetime.fromisoformat(raw) if parsed.tzinfo is None: raise ValueError("object creation time must include UTC offset") - return parsed.astimezone(timezone.utc) + return parsed.astimezone(UTC) def purge(client: ObjectClient, *, now: datetime) -> dict[str, object]: @@ -118,9 +117,11 @@ def _request(self, method: str, url: str) -> dict[str, object] | None: if method == "DELETE": return None data = json.load(response) - if not isinstance(data, dict): + # JSON object keys are strings; their values remain untrusted objects. + object_data = cast(dict[str, object], data) if isinstance(data, dict) else None + if object_data is None: raise ValueError("invalid GCS response") - return data + return object_data def list_all(self) -> tuple[StoredObject, ...]: found: list[StoredObject] = [] @@ -134,13 +135,15 @@ def list_all(self) -> tuple[StoredObject, ...]: data = self._request("GET", self._storage_api + "?" + urlencode(query)) assert data is not None items = data.get("items", []) - if not isinstance(items, list): + object_items = cast(list[object], items) if isinstance(items, list) else None + if object_items is None: raise ValueError("invalid GCS object list") - for raw in items: - if not isinstance(raw, dict): + for raw in object_items: + metadata = cast(Mapping[object, object], raw) if isinstance(raw, dict) else None + if metadata is None: raise ValueError("invalid GCS object metadata") name, generation, created = ( - raw.get("name"), raw.get("generation"), raw.get("timeCreated") + metadata.get("name"), metadata.get("generation"), metadata.get("timeCreated") ) if not isinstance(name, str) or not name or not isinstance(created, str): raise ValueError("invalid GCS object metadata") @@ -170,10 +173,9 @@ def delete_generation(self, obj: StoredObject) -> None: def main() -> None: bucket = bucket_from_environment(os.environ) - result = purge(GcsJsonClient(bucket), now=datetime.now(timezone.utc)) + result = purge(GcsJsonClient(bucket), now=datetime.now(UTC)) print(json.dumps(result, sort_keys=True)) if __name__ == "__main__": main() - diff --git a/c3r/serve.py b/c3r/serve.py index be05dd5..a181e63 100644 --- a/c3r/serve.py +++ b/c3r/serve.py @@ -52,9 +52,10 @@ def load_host_builder(reference: str) -> HostBuilder: raise ValueError("C3R_HOST_ENTRYPOINT must be module:function") module = importlib.import_module(module_name) builder = getattr(module, attribute) - if not callable(builder): + callable_builder = cast(HostBuilder, builder) if callable(builder) else None + if callable_builder is None: raise ValueError("C3R_HOST_ENTRYPOINT is not callable") - return cast(HostBuilder, builder) + return callable_builder def build_servers( diff --git a/c3r/staging_host.py b/c3r/staging_host.py index 06b3655..aeb169f 100644 --- a/c3r/staging_host.py +++ b/c3r/staging_host.py @@ -18,7 +18,6 @@ from .telemetry.ephemeral import EphemeralTraceSink from .verifier_firewall import VerifierDecision, VerifierFirewall, VerifierPolicy - EphemeralStagingSink = EphemeralTraceSink diff --git a/c3r/state_compiler.py b/c3r/state_compiler.py index 0bc7d33..b52940b 100644 --- a/c3r/state_compiler.py +++ b/c3r/state_compiler.py @@ -29,13 +29,16 @@ def __post_init__(self) -> None: raise ValueError("all state limits must be positive") +_DEFAULT_LIMITS = StateLimits() + + class StateCompiler: """Compile runtime history without inventing or dropping consequential evidence.""" def __init__( self, minimum_confidence: float = 0.5, - limits: StateLimits = StateLimits(), + limits: StateLimits = _DEFAULT_LIMITS, ) -> None: if not 0.0 <= minimum_confidence <= 1.0: raise ValueError("minimum_confidence must be between zero and one") diff --git a/c3r/state_schema.py b/c3r/state_schema.py index f12af4f..4baea3e 100644 --- a/c3r/state_schema.py +++ b/c3r/state_schema.py @@ -2,10 +2,9 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass, field from enum import StrEnum -from typing import Mapping - SCHEMA_VERSION = "c3r.state.v1" @@ -53,16 +52,18 @@ class RawState: available_action_families: tuple[ActionFamily, ...] = () tool_summary: tuple[str, ...] = () model_inventory: tuple[str, ...] = () - budget: Mapping[str, float] = field(default_factory=dict) - runtime_summary: Mapping[str, str | float | bool] = field(default_factory=dict) - risk: Mapping[str, str | float | bool] = field(default_factory=dict) + budget: Mapping[str, float] = field(default_factory=dict[str, float]) + runtime_summary: Mapping[str, str | float | bool] = field( + default_factory=dict[str, str | float | bool] + ) + risk: Mapping[str, str | float | bool] = field(default_factory=dict[str, str | float | bool]) reversibility: str = "unknown" approval_required: bool = False data_boundary: str = "unknown" rollback_state: str | None = None ambiguity: float = 0.0 consequence: str = "low" - provenance: Mapping[str, Provenance] = field(default_factory=dict) + provenance: Mapping[str, Provenance] = field(default_factory=dict[str, Provenance]) @dataclass(frozen=True, slots=True) diff --git a/c3r/system_one/clm_adapter.py b/c3r/system_one/clm_adapter.py index 1f11cb1..0c91856 100644 --- a/c3r/system_one/clm_adapter.py +++ b/c3r/system_one/clm_adapter.py @@ -18,6 +18,7 @@ from ..http_transport import NoRedirectHandler from ..state_schema import CompiledState +from ..telemetry.invocations import transport_attempt from .question_registry import TypedQuestion UPSTREAM_CLM_COMMIT = "bb42c6c5bf914fd449bed2f6ca65be80602cb1f7" @@ -120,18 +121,21 @@ def _rank( remaining = min(remaining, deadline - time.monotonic()) if remaining <= 0: raise TimeoutError("CLM decision deadline exceeded") + transport_attempt("system_one") response = self.transport(payload) if self.transport is not None else self._post(payload, remaining) if response.get("model") != self.served_model: raise ValueError("CLM served model does not match the requested model") ranked = response.get("ranked") - if not isinstance(ranked, list) or len(ranked) != len(options): + if not isinstance(ranked, list) or len(cast(list[object], ranked)) != len(options): raise ValueError("CLM returned an incomplete ranking") + ranking_items = cast(list[object], ranked) probabilities: dict[str, float] = {} - for item in ranked: - if not isinstance(item, dict): + for item in ranking_items: + ranking_item = cast(Mapping[object, object], item) if isinstance(item, dict) else None + if ranking_item is None: raise ValueError("CLM returned an invalid ranking item") - candidate = item.get("candidate") - probability = item.get("prob") + candidate = ranking_item.get("candidate") + probability = ranking_item.get("prob") if ( not isinstance(candidate, str) or candidate not in options @@ -166,6 +170,7 @@ def _post(self, payload: Mapping[str, object], timeout: float) -> Mapping[str, o if len(body) > _MAX_RESPONSE_BYTES: raise ValueError("CLM response exceeds the allowed size") parsed = json.loads(body) - if not isinstance(parsed, dict): + object_response = cast(Mapping[str, object], parsed) if isinstance(parsed, dict) else None + if object_response is None: raise ValueError("CLM returned a non-object response") - return cast(Mapping[str, object], parsed) + return object_response diff --git a/c3r/system_one/fast_path.py b/c3r/system_one/fast_path.py index d84f6df..c0dcb64 100644 --- a/c3r/system_one/fast_path.py +++ b/c3r/system_one/fast_path.py @@ -2,10 +2,10 @@ from __future__ import annotations -from dataclasses import dataclass import math import time from collections.abc import Mapping +from dataclasses import dataclass from typing import Protocol from ..state_schema import CompiledState @@ -15,9 +15,14 @@ class TypedInferenceAdapter(Protocol): - model_id: str - revision: str - provider: str + @property + def model_id(self) -> str: ... + + @property + def revision(self) -> str: ... + + @property + def provider(self) -> str: ... def predict( self, state: CompiledState, questions: tuple[TypedQuestion, ...], diff --git a/c3r/system_one/inference.py b/c3r/system_one/inference.py index ad38a07..f5a4a66 100644 --- a/c3r/system_one/inference.py +++ b/c3r/system_one/inference.py @@ -17,15 +17,17 @@ def _text(value: object, limit: int = 1024) -> str: def _mapping(value: object) -> Mapping[str, object]: - if not isinstance(value, dict): + mapping = cast(Mapping[str, object], value) if isinstance(value, dict) else None + if mapping is None: raise ValueError("object required") - return cast(Mapping[str, object], value) + return mapping def _list(value: object) -> list[object]: - if not isinstance(value, list): + sequence = cast(list[object], value) if isinstance(value, list) else None + if sequence is None: raise ValueError("array required") - return cast(list[object], value) + return sequence class SystemOneInference: diff --git a/c3r/system_one/laya_adapter.py b/c3r/system_one/laya_adapter.py index 8149d46..a64422d 100644 --- a/c3r/system_one/laya_adapter.py +++ b/c3r/system_one/laya_adapter.py @@ -2,9 +2,9 @@ from __future__ import annotations +import re from collections.abc import Callable, Mapping from dataclasses import dataclass -import re from ..state_schema import CompiledState from .question_registry import TypedQuestion diff --git a/c3r/system_one/laya_backend.py b/c3r/system_one/laya_backend.py index 8f216ef..e280aed 100644 --- a/c3r/system_one/laya_backend.py +++ b/c3r/system_one/laya_backend.py @@ -2,9 +2,10 @@ from __future__ import annotations +import math from collections.abc import Callable, Mapping from dataclasses import asdict, dataclass -import math +from importlib import import_module from typing import Protocol, cast from ..state_schema import CompiledState @@ -30,6 +31,16 @@ def predict( AgentLoader = Callable[[str, str | None], LayaAgent] +class _SnapshotDownloader(Protocol): + def __call__( + self, *, repo_id: str, revision: str, allow_patterns: list[str] + ) -> str: ... + + +class _SdkAgentLoader(Protocol): + def __call__(self, path: str, *, device: str | None) -> LayaAgent: ... + + def _resolve_metadata(model_id: str, revision: str) -> HubModelMetadata: try: from huggingface_hub import HfApi @@ -37,7 +48,7 @@ def _resolve_metadata(model_id: str, revision: str) -> HubModelMetadata: raise RuntimeError("install the 'laya' extra to use the live Laya backend") from error info = HfApi().model_info(repo_id=model_id, revision=revision) card_data = info.card_data - if hasattr(card_data, "to_dict"): + if card_data is not None and hasattr(card_data, "to_dict"): card_data = card_data.to_dict() license_name = str((card_data or {}).get("license", "")) return HubModelMetadata(sha=str(info.sha), license=license_name) @@ -45,9 +56,15 @@ def _resolve_metadata(model_id: str, revision: str) -> HubModelMetadata: def _fetch_snapshot(model_id: str, revision: str) -> str: try: - from huggingface_hub import snapshot_download + hub = import_module("huggingface_hub") except ImportError as error: raise RuntimeError("install the 'laya' extra to use the live Laya backend") from error + download: object = getattr(hub, "snapshot_download", None) + if not callable(download): + raise TypeError("installed Hub SDK has no snapshot_download callable") + # Describe only the documented keyword subset used here, not the SDK's + # unrelated optional parameters. Callability is checked, not attested. + snapshot_download = cast(_SnapshotDownloader, download) return str( snapshot_download( repo_id=model_id, @@ -66,10 +83,16 @@ def _fetch_snapshot(model_id: str, revision: str) -> str: def _load_agent(path: str, device: str | None) -> LayaAgent: try: - import laya + laya = import_module("laya") except ImportError as error: raise RuntimeError("install the 'laya' extra to use the live Laya backend") from error - return cast(LayaAgent, laya.load(path, device=device)) + load: object = getattr(laya, "load", None) + if not callable(load): + raise TypeError("installed Laya SDK has no load callable") + # The optional SDK contract is load(path, device=...) -> prediction agent; + # this annotation does not grant its predictions authority or calibration. + loader = cast(_SdkAgentLoader, load) + return loader(path, device=device) class PinnedLayaBackend: diff --git a/c3r/telemetry/ephemeral.py b/c3r/telemetry/ephemeral.py index 37de5d3..75764fd 100644 --- a/c3r/telemetry/ephemeral.py +++ b/c3r/telemetry/ephemeral.py @@ -3,7 +3,7 @@ from __future__ import annotations from .trace import DecisionTrace -from .trace_ledger import LedgerRecord, _record_hash, canonical_trace_json +from .trace_ledger import LedgerRecord, canonical_trace_json, record_hash class EphemeralTraceSink: @@ -12,4 +12,4 @@ class EphemeralTraceSink: def append(self, trace: DecisionTrace) -> LedgerRecord: canonical = canonical_trace_json(trace) genesis = "0" * 64 - return LedgerRecord(genesis, _record_hash(genesis, canonical), canonical) + return LedgerRecord(genesis, record_hash(genesis, canonical), canonical) diff --git a/c3r/telemetry/governed_store.py b/c3r/telemetry/governed_store.py index 2bd3fc0..6d6c94a 100644 --- a/c3r/telemetry/governed_store.py +++ b/c3r/telemetry/governed_store.py @@ -11,17 +11,17 @@ import re import sqlite3 import stat +from collections.abc import Callable from dataclasses import asdict, dataclass -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from math import isfinite from pathlib import Path from threading import Lock -from typing import Callable +from typing import Self from .ledger_anchor import LedgerHead, capture_head from .trace import DecisionTrace -from .trace_ledger import LedgerRecord, _record_hash - +from .trace_ledger import LedgerRecord, record_hash RETENTION_DAYS = 30 _TOKEN = re.compile(r"[A-Za-z0-9_.:-]{1,128}\Z") @@ -29,6 +29,10 @@ _GENESIS = "0" * 64 +def _finite_number(value: object) -> bool: + return isinstance(value, (int, float)) and isfinite(value) + + @dataclass(frozen=True, slots=True) class SourceGrant: source_id: str @@ -81,7 +85,7 @@ def _validate_trace(trace: DecisionTrace) -> None: raise ValueError("trace failed redaction schema") elif isinstance(value, bool): pass - elif not isinstance(value, (int, float)) or not isfinite(value): + elif not _finite_number(value): raise ValueError("trace failed redaction schema") @@ -97,7 +101,7 @@ def __init__( path: Path, *, grants: tuple[SourceGrant, ...], - clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc), + clock: Callable[[], datetime] = lambda: datetime.now(UTC), ) -> None: if not path.parent.is_dir() or path.is_symlink(): raise ValueError("store parent must exist and path must not be a symlink") @@ -139,7 +143,7 @@ def __init__( self._db.close() raise ValueError("governed trace hash chain is invalid") - def __enter__(self) -> GovernedTraceStore: + def __enter__(self) -> Self: return self def __exit__(self, *_args: object) -> None: @@ -188,7 +192,7 @@ def _verified_head(self) -> tuple[int, str] | None: return None except (ValueError, TypeError, KeyError): return None - if canonical != payload or _record_hash(prior, payload) != digest: + if canonical != payload or record_hash(prior, payload) != digest: return None previous = digest return sequence, previous @@ -244,7 +248,7 @@ def append(self, trace: DecisionTrace, *, source_id: str, task_id: str) -> Ledge "SELECT record_hash FROM records ORDER BY sequence DESC LIMIT 1" ).fetchone() previous = row[0] if row is not None else checkpoint - record = LedgerRecord(previous, _record_hash(previous, payload), payload) + record = LedgerRecord(previous, record_hash(previous, payload), payload) self._db.execute( "INSERT INTO records (run_id, collected_at, previous_hash, " "record_hash, canonical_json) VALUES (?, ?, ?, ?, ?)", @@ -287,6 +291,12 @@ def purge_expired(self) -> int: self._db.execute("VACUUM") return len(expired) + def validate_task(self, *, source_id: str, task_id: str) -> None: + """Validate a host-selected binding without exposing the source registry.""" + grant = self._grants.get(source_id) + if grant is None or task_id not in grant.task_ids: + raise ValueError("unapproved source or task") + def close(self) -> None: with self._lock: self._db.close() @@ -300,9 +310,7 @@ class BoundGovernedTraceSink: """ def __init__(self, store: GovernedTraceStore, *, source_id: str, task_id: str) -> None: - grant = store._grants.get(source_id) - if grant is None or task_id not in grant.task_ids: - raise ValueError("unapproved source or task") + store.validate_task(source_id=source_id, task_id=task_id) self._store = store self._source_id = source_id self._task_id = task_id diff --git a/c3r/telemetry/invocations.py b/c3r/telemetry/invocations.py new file mode 100644 index 0000000..e07ebef --- /dev/null +++ b/c3r/telemetry/invocations.py @@ -0,0 +1,34 @@ +"""Request-local transport-attempt counters, not GPU work or billing estimates.""" +from collections.abc import Generator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar +from typing import cast + +_counts: ContextVar[dict[str, int] | None] = ContextVar("c3r_invocations", default=None) + + +@contextmanager +def capture_invocations() -> Generator[None]: + token = _counts.set({"system_one_invocations": 0, "system_two_invocations": 0}) + try: + yield + finally: + _counts.reset(token) + + +def transport_attempt(system: str) -> None: + counts = _counts.get() + if counts is not None: + counts[system + "_invocations"] += 1 + + +def with_invocations(value: Mapping[str, object]) -> dict[str, object]: + result = dict(value) + counts = _counts.get() + if counts is not None: + previous = result.get("c3r") + metadata: dict[str, object] = dict(cast(Mapping[str, object], previous)) if isinstance(previous, dict) else {} + metadata.update(counts) + metadata["invocation_basis"] = "adapter_transport_attempts" + result["c3r"] = metadata + return result diff --git a/c3r/telemetry/ledger_anchor.py b/c3r/telemetry/ledger_anchor.py index 2ac6296..faa8476 100644 --- a/c3r/telemetry/ledger_anchor.py +++ b/c3r/telemetry/ledger_anchor.py @@ -9,10 +9,9 @@ import json import re +from collections.abc import Callable from dataclasses import asdict, dataclass -from datetime import datetime, timezone -from typing import Callable - +from datetime import UTC, datetime _HASH = re.compile(r"[0-9a-f]{64}\Z") @@ -33,7 +32,8 @@ def __post_init__(self) -> None: instant = datetime.fromisoformat(self.captured_at) except ValueError as error: raise ValueError("invalid capture timestamp") from error - if instant.tzinfo is None or instant.utcoffset().total_seconds() != 0: + offset = instant.utcoffset() + if instant.tzinfo is None or offset is None or offset.total_seconds() != 0: raise ValueError("capture timestamp must be UTC") def payload(self) -> bytes: @@ -51,7 +51,7 @@ class SignedLedgerAnchor: def capture_head( *, sequence: int, record_hash: str, policy_version: str, - clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc), + clock: Callable[[], datetime] = lambda: datetime.now(UTC), ) -> LedgerHead: return LedgerHead(sequence, record_hash, clock().isoformat(), policy_version) diff --git a/c3r/telemetry/sqlite_ledger.py b/c3r/telemetry/sqlite_ledger.py index 5e171c0..e1b19f4 100644 --- a/c3r/telemetry/sqlite_ledger.py +++ b/c3r/telemetry/sqlite_ledger.py @@ -2,14 +2,14 @@ from __future__ import annotations -import sqlite3 import os +import sqlite3 import stat from pathlib import Path from threading import Lock from .trace import DecisionTrace -from .trace_ledger import LedgerRecord, TraceLedger, _record_hash, canonical_trace_json +from .trace_ledger import LedgerRecord, TraceLedger, canonical_trace_json, record_hash class SqliteTraceLedger: @@ -59,7 +59,7 @@ def append(self, trace: DecisionTrace) -> LedgerRecord: "SELECT record_hash FROM records ORDER BY sequence DESC LIMIT 1" ).fetchone() previous = row[0] if row is not None else "0" * 64 - record = LedgerRecord(previous, _record_hash(previous, canonical), canonical) + record = LedgerRecord(previous, record_hash(previous, canonical), canonical) self._db.execute( "INSERT INTO records (previous_hash, record_hash, canonical_json) " "VALUES (?, ?, ?)", diff --git a/c3r/telemetry/trace.py b/c3r/telemetry/trace.py index c9923b4..887672c 100644 --- a/c3r/telemetry/trace.py +++ b/c3r/telemetry/trace.py @@ -2,8 +2,8 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass -from typing import Mapping @dataclass(frozen=True, slots=True) diff --git a/c3r/telemetry/trace_ledger.py b/c3r/telemetry/trace_ledger.py index 816c8f2..9ceb311 100644 --- a/c3r/telemetry/trace_ledger.py +++ b/c3r/telemetry/trace_ledger.py @@ -19,12 +19,17 @@ class LedgerRecord: canonical_json: str -def _record_hash(previous_hash: str, canonical_json: str) -> str: +def record_hash(previous_hash: str, canonical_json: str) -> str: return hashlib.sha256( (previous_hash + "\n" + canonical_json).encode("utf-8") ).hexdigest() +# Compatibility for callers of the historical private name. Shared sinks use +# the public canonical hash interface rather than reaching into this module. +_record_hash = record_hash + + def canonical_trace_json(trace: DecisionTrace) -> str: return json.dumps( asdict(trace), @@ -51,7 +56,7 @@ def append(self, trace: DecisionTrace) -> LedgerRecord: canonical = canonical_trace_json(trace) with self._lock: previous = self._records[-1].record_hash if self._records else _GENESIS_HASH - record = LedgerRecord(previous, _record_hash(previous, canonical), canonical) + record = LedgerRecord(previous, record_hash(previous, canonical), canonical) self._records.append(record) return record @@ -98,7 +103,7 @@ def verify(records: tuple[LedgerRecord, ...]) -> bool: return False if canonical != record.canonical_json: return False - if _record_hash(previous, canonical) != record.record_hash: + if record_hash(previous, canonical) != record.record_hash: return False previous = record.record_hash return True diff --git a/c3r/verifier_firewall.py b/c3r/verifier_firewall.py index decde7b..cc6873a 100644 --- a/c3r/verifier_firewall.py +++ b/c3r/verifier_firewall.py @@ -21,7 +21,7 @@ class VerifierDecision: @dataclass(frozen=True, slots=True) class VerifierPolicy: default_verifier: str - by_risk: Mapping[RiskClass, str] = field(default_factory=dict) + by_risk: Mapping[RiskClass, str] = field(default_factory=dict[RiskClass, str]) version: str = "policy-v1" def select(self, candidate: ActionCandidate) -> str: diff --git a/docs/api-key-recovery.md b/docs/api-key-recovery.md new file mode 100644 index 0000000..31b1a2f --- /dev/null +++ b/docs/api-key-recovery.md @@ -0,0 +1,62 @@ +# Operator-local credential recovery + +The key-management CLI can now create a credential-only SQLite snapshot and restore +it into a **new** database. This implements the local backup/restore boundary, not +deployed encryption, a backup schedule, copy deletion, or an actual production drill. + +## Safety contract + +- Copy organizations, project rate limits and hash-only key records in one SQLite + read transaction. Do not copy quota windows, usage records or request audit history. +- Reject missing sources, linked paths and existing destination files. POSIX private + directory/file checks remain in force; Windows ACLs require deployment verification. +- Return the completed snapshot's SHA256, counts and scope, never plaintext keys. +- Require the approved snapshot SHA256 on restore and a versioned snapshot marker. + The hash must come from the approved receipt, not be recomputed to bless a changed file. +- Restore only a bounded, standalone snapshot (at most 64 MiB), rejecting WAL, SHM + and journal sidecars. Deserialize the exact hash-verified bytes into read-only + in-memory SQLite; never reopen the source pathname to obtain recovery rows. +- Backup opens the existing database read-only, validates its credential schema and + uses one read transaction. An empty or wrong source is an error, not a new database. + Hard-linked files and redirected/junction paths are rejected at the recovery boundary. +- Revoke **all restored keys**. An older backup cannot establish revocations made + after it was taken. Issue replacement scoped keys through the secure delivery + workflow before reopening ingress; no option silently reactivates snapshot keys. +- Preserve the original database and snapshot. Recovery never overwrites the live + database. An unsuccessful copy may leave a reserved empty database; quarantine it + and use a new path after diagnosis, rather than reusing it as a valid snapshot. + +## Commands (private operator terminal only) + +Both source and destination must be absolute paths in approved private directories. +The destination must not exist. Provision and verify encrypted storage and its IAM +before copying real credentials; a SQLite file is not itself encrypted. + +```sh +python -m c3r.key_management --database /private/live/access.sqlite3 \ + backup --destination /private/encrypted-backups/access-snapshot.sqlite3 + +python -m c3r.key_management --database /private/encrypted-backups/access-snapshot.sqlite3 \ + restore --destination /private/recovery/access.sqlite3 \ + --expected-sha256 APPROVED_SNAPSHOT_SHA256 +``` + +Keep the backup hash and receipt in the approved private evidence channel. Backup +records contain credential hashes/salts and tenant identifiers: they are sensitive +even without plaintext keys. Never upload them to a public repository or log their +contents. Snapshot credentials preserve their absolute expiration and scope; restores +add revocation and require replacement keys rather than extending old expiration. + +With ingress disabled, verify project isolation, expected credential counts, refusal +of old keys and successful authentication of securely delivered replacement keys. +Switch the configured database only through the reviewed service recovery procedure. +Quota windows restart in the new database; do not use recovery to evade rate limits. + +## Remaining deployment evidence + +The production gate still requires actual encrypted-volume/IAM readbacks, encrypted +backup destinations, a scheduled snapshot job, inventory of every copy/version, +retention and deletion evidence, tested restoration and key delivery, and a live +rollback drill. Receipts deliberately report `encryption_verified=false` and +`backup_deletion_verified=false`. This implementation does not enable production, +research collection, MC-1 or `/v1/c3r/execute`. diff --git a/docs/api-qualification-followon.md b/docs/api-qualification-followon.md new file mode 100644 index 0000000..f0d0e53 --- /dev/null +++ b/docs/api-qualification-followon.md @@ -0,0 +1,52 @@ +# API qualification follow-on candidate + +This candidate is separate from frozen PR #7 source `7d3a93f`. It is not deployed, +image-qualified or a production approval. External execution remains HTTP 501; +tenant identity remains scoped-key-derived and caller override headers rejected. + +The gateway bounds accepted connection workers to 32 by default, before parsing +headers or reading bodies. `C3RIngressServer(max_connections=...)` permits an +operator-tested integer limit from 1 to 4096. Saturation closes the newly accepted +transport without allocating a worker: no identity has been authenticated, so this +is not a tenant usage event or an HTTP 429 guarantee. Existing per-route and global +dispatch limits remain separate. Idle transports time out after five seconds; +these limits are not measured GPU capacity or distributed admission. + +Authenticated refusals and observed pre-header disconnects produce at most one +payload-free usage outcome (499 for an observed disconnect). Unknown identities +are not invented. If storage fails, requests fail closed, but durable accounting +cannot be claimed for an unavailable database. Connection refusal is not recorded +as billable model work. + +CLM ranking, structured ProviderAdapter deliberation and OpenAI-compatible text generation count request-local transport +attempts at their adapter dispatch boundaries, including attempted failures. SSE +terminal responses carry the same counters. `invocation_basis` is explicitly +`adapter_transport_attempts`: counts are not successful GPU executions, completed +generations, cost allocation, or model labels inferred from routing. Uninstrumented +providers, readiness probes outside request capture and absent upstream metadata +remain unknown; the counters do not cover arbitrary provider internals or retries. +Zero means no instrumented transport attempt in that captured backend request. + +Operator key revocation succeeds only for a key in the selected organization and +project. A missing or cross-project key returns the CLI's sanitized failure and +does not emit a successful revocation audit event. This guards against falsely +certifying a rotation/revocation drill; it is not an authenticated public admin API. + +The trusted operator CLI adds project-scoped bounded logical deletion: + +```text +python -m c3r.key_management --database purge \ + --tenant --project \ + --retention-seconds 2592000 --limit 1000 +``` + +Retention is bounded to 60 seconds through 30 days. Each invocation deletes at most +the requested limit per usage/audit table, only for the selected project and only +rows older than its cutoff. Repeat through an approved scheduler until backlog is +cleared; no scheduler is installed by this change. Keys/projects are not deleted. +SQLite secure-delete is enabled for this purge, but the receipt explicitly does +not claim physical erasure or backup/copy deletion. Deployments must independently +inventory and delete database snapshots, replicas, exported metadata, journals and +backups within their approved retention policy. Encryption, IAM, backup restore, +TLS, alerts, external monitoring and live no-payload logging remain deployment +qualification gates, not properties established by these local fixture tests. diff --git a/docs/strict-static-release-repair.md b/docs/strict-static-release-repair.md new file mode 100644 index 0000000..97e91ce --- /dev/null +++ b/docs/strict-static-release-repair.md @@ -0,0 +1,19 @@ +# Strict static release repair + +Review baseline: `71bd6d30838f6fe7eafd9af7840b3fba31ac5bd4`. + +The standalone release requires full strict static checks, not merely clean changed files. The baseline produced 461 Pyright diagnostics and 56 configured Ruff findings. Preserve the existing Pyright strict mode, full `c3r`/`tests` include set, Ruff configuration and runtime assertions: do not suppress diagnostics, introduce fake optional dependency stubs, remove tests or use blanket `Any`/unchecked schema casts. + +Repair requirements: + +- Type dataclass default factories and existing fixture/server/callback interfaces accurately. +- Keep untrusted JSON values as objects until actual structural validation, and preserve rejection messages and exception contracts. +- Express inference adapter metadata as read-only properties; implementations must not gain mutation or authority. +- Keep optional Laya dependencies lazy, missing-extra failures and immutable revision/license checks. Narrow only the documented SDK callable interfaces; injected fixture success is not live SDK qualification. +- Give shared canonical hashing and task admission explicit public interfaces without changing hash bytes, grant semantics or enabling collection. Preserve the historical private hash alias for compatibility. +- Preserve recommendation-only rejection, transport failure/stream cancellation tests, project isolation, generation-specific retention and timezone rejection. Use public transport boundaries instead of private socket/helper access where practical. +- Fix mechanical import, timezone, immutable-default and fixture-loop issues without changing supported Python versions or release scope. + +Acceptance: full strict Pyright and configured Ruff pass, full existing suite including official Python/JS SDK fixtures passes with zero skips, and separate Standards/Spec review finds no unresolved correctness or requirements defects. + +These repairs change the API source candidate. Existing exact-head `71bd6d3` image configurations and evidence remain historical; they cannot qualify the new candidate. Rebind API/retention qualification after code review and final source freeze. DeepSeek build-context equality must be checked before reusing its preserved artifact. No local static/test result clears host, CVE, cloud, registry, TLS, canary or Avori gates. diff --git a/scripts/build_decisionmix_preview.py b/scripts/build_decisionmix_preview.py index 8c065b1..9783a87 100644 --- a/scripts/build_decisionmix_preview.py +++ b/scripts/build_decisionmix_preview.py @@ -2,8 +2,8 @@ from __future__ import annotations -from pathlib import Path import sys +from pathlib import Path REPOSITORY_ROOT = Path(__file__).resolve().parents[1] if str(REPOSITORY_ROOT) not in sys.path: @@ -12,7 +12,6 @@ from c3r.decisionmix import DecisionMixBuilder, DecisionMixRecord, SourceProvenance from c3r.state_schema import ActionFamily - GENERATOR_REVISION = "0000000000000000000000000000000000000001" OUTPUT = REPOSITORY_ROOT / "data" / "decisionmix-v1-preview" diff --git a/scripts/run_trace_control_dry_run.py b/scripts/run_trace_control_dry_run.py index 28000de..95b2478 100644 --- a/scripts/run_trace_control_dry_run.py +++ b/scripts/run_trace_control_dry_run.py @@ -8,7 +8,7 @@ import json import sys from collections.abc import Callable -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from pathlib import Path from tempfile import TemporaryDirectory @@ -19,8 +19,7 @@ from c3r.telemetry.governed_store import GovernedTraceStore, SourceGrant from c3r.telemetry.trace import DecisionTrace - -START = datetime(2026, 9, 23, 0, 0, tzinfo=timezone.utc) +START = datetime(2026, 9, 23, 0, 0, tzinfo=UTC) SOURCE_ID = "c3r_fixture_review" TASK_ID = "fixture_task_001" @@ -58,38 +57,37 @@ def run() -> dict[str, object]: task_ids=frozenset({TASK_ID}), rights_attested=True, ) clock = [START] - with TemporaryDirectory(prefix="c3r-trace-review-") as directory: - with GovernedTraceStore( - Path(directory) / "trace.sqlite3", grants=(grant,), clock=lambda: clock[0], - ) as store: - first = store.append(_trace("fixture_run_001"), source_id=SOURCE_ID, task_id=TASK_ID) - checks = { - "approved_fixture_admitted": len(store.records()) == 1, - "unapproved_source_rejected": _rejects(lambda: store.append( - _trace("fixture_run_002"), source_id="customer_logs", task_id=TASK_ID, - ), "unapproved source or task"), - "free_text_rejected": _rejects(lambda: store.append( - _trace("fixture_run_003", task_outcome={"status": "email me at a@example.com"}), - source_id=SOURCE_ID, task_id=TASK_ID, - ), "redaction"), - "artifact_reference_rejected": _rejects(lambda: store.append( - _trace("fixture_run_004", artifact_refs=("private_artifact",)), - source_id=SOURCE_ID, task_id=TASK_ID, - ), "redaction"), - } - checks["rejected_rows_not_persisted"] = len(store.records()) == 1 - clock[0] = START + timedelta(days=31) - checks["overdue_purge_blocks_collection"] = _rejects(lambda: store.append( - _trace("fixture_run_005"), source_id=SOURCE_ID, task_id=TASK_ID, - ), "retention purge overdue") - checks["local_expired_row_purged"] = store.purge_expired() == 1 - checks["checkpoint_chain_verifies"] = store.verify() and not store.records() - second = store.append( - _trace("fixture_run_006"), source_id=SOURCE_ID, task_id=TASK_ID, - ) - checks["post_purge_chain_continues"] = ( - second.previous_hash == first.record_hash and store.verify() - ) + with TemporaryDirectory(prefix="c3r-trace-review-") as directory, GovernedTraceStore( + Path(directory) / "trace.sqlite3", grants=(grant,), clock=lambda: clock[0], + ) as store: + first = store.append(_trace("fixture_run_001"), source_id=SOURCE_ID, task_id=TASK_ID) + checks = { + "approved_fixture_admitted": len(store.records()) == 1, + "unapproved_source_rejected": _rejects(lambda: store.append( + _trace("fixture_run_002"), source_id="customer_logs", task_id=TASK_ID, + ), "unapproved source or task"), + "free_text_rejected": _rejects(lambda: store.append( + _trace("fixture_run_003", task_outcome={"status": "email me at a@example.com"}), + source_id=SOURCE_ID, task_id=TASK_ID, + ), "redaction"), + "artifact_reference_rejected": _rejects(lambda: store.append( + _trace("fixture_run_004", artifact_refs=("private_artifact",)), + source_id=SOURCE_ID, task_id=TASK_ID, + ), "redaction"), + } + checks["rejected_rows_not_persisted"] = len(store.records()) == 1 + clock[0] = START + timedelta(days=31) + checks["overdue_purge_blocks_collection"] = _rejects(lambda: store.append( + _trace("fixture_run_005"), source_id=SOURCE_ID, task_id=TASK_ID, + ), "retention purge overdue") + checks["local_expired_row_purged"] = store.purge_expired() == 1 + checks["checkpoint_chain_verifies"] = store.verify() and not store.records() + second = store.append( + _trace("fixture_run_006"), source_id=SOURCE_ID, task_id=TASK_ID, + ) + checks["post_purge_chain_continues"] = ( + second.previous_hash == first.record_hash and store.verify() + ) return { "evidence_kind": "non_sensitive_local_fixture_dry_run", "live_trace_collection_enabled": False, diff --git a/tests/test_api_access_http.py b/tests/test_api_access_http.py index c2878bb..adebdc3 100644 --- a/tests/test_api_access_http.py +++ b/tests/test_api_access_http.py @@ -138,12 +138,38 @@ def test_key_scope_blocks_generation_with_standard_error_and_request_id(self) -> self.assertEqual((status, error["type"]), (403, "permission_error")) self.assertEqual(error["request_id"], request_id) + def test_authenticated_refusal_has_one_payload_free_outcome(self) -> None: + status, _, request_id = self.call("/v1/responses", method="POST", payload={ + "model": "c3r-core", "input": "PRIVATE_REFUSAL_MARKER"}) + rows = self.store.project_usage("tenant-a", "project-a") + self.assertEqual(len(rows), 1) + self.assertEqual((rows[0]["request_id"], rows[0]["status"]), (request_id, status)) + self.assertEqual(status, 403) + self.assertIsNone(rows[0]["system_one_invocations"]) + self.assertNotIn("PRIVATE_REFUSAL_MARKER", json.dumps(rows)) + def test_revoked_and_expired_keys_cannot_infer(self) -> None: self.store.revoke_key("tenant-a", "project-a", self.key.key_id) self.assertEqual(self.call("/v1/models")[0], 401) expired = self.store.issue_key("tenant-a", "project-a", {"models:read"}, expires_at=1) self.assertEqual(self.call("/v1/models", key=expired.secret)[0], 401) + def test_operator_revoke_cannot_claim_success_for_another_projects_key(self) -> None: + command = [sys.executable, "-m", "c3r.key_management", "--database", + str(self.store.path), "revoke", "--tenant", "tenant-a", + "--key-id", self.key.key_id, "--project"] + wrong_project = subprocess.run(command + ["other-project"], capture_output=True, + text=True, timeout=5, check=False) + self.assertEqual(wrong_project.returncode, 1) + self.assertIn("error", json.loads(wrong_project.stdout)) + self.assertNotIn(self.key.secret, wrong_project.stdout) + self.assertEqual(self.call("/v1/models")[0], 200) + own_project = subprocess.run(command + ["project-a"], capture_output=True, + text=True, timeout=5, check=False) + self.assertEqual(own_project.returncode, 0) + self.assertEqual(json.loads(own_project.stdout), {"status": "revoked"}) + self.assertEqual(self.call("/v1/models")[0], 401) + def test_project_rate_limit_is_shared_by_its_keys_not_other_tenants(self) -> None: self.store.create_project("tenant-b", "project-a", rpm=1) first = self.store.issue_key("tenant-b", "project-a", {"models:read"}) @@ -185,6 +211,27 @@ def test_usage_cli_reports_actual_counts_without_payloads_or_fabricated_gpu_cost self.assertNotIn("PRIVATE_", result.stdout) self.assertNotIn(key.secret, self.store.path.read_bytes().decode("latin1")) + def test_operator_metadata_purge_removes_only_aged_rows_in_selected_project(self) -> None: + self.store.clock = lambda: 1.0 + self.assertEqual(self.call("/v1/models")[0], 200) + self.store.create_project("tenant-b", "project-a") + other = self.store.issue_key("tenant-b", "project-a", {"models:read"}) + self.assertEqual(self.call("/v1/models", key=other.secret)[0], 200) + self.store.clock = time.time + self.assertEqual(self.call("/v1/models")[0], 200) + result = subprocess.run([sys.executable, "-m", "c3r.key_management", "--database", + str(self.store.path), "purge", "--tenant", "tenant-a", + "--project", "project-a", "--retention-seconds", "86400", + "--limit", "10"], capture_output=True, text=True, timeout=5, + check=False) + self.assertEqual(result.returncode, 0, result.stderr) + proof = json.loads(result.stdout) + self.assertEqual((proof["usage_deleted"], proof["audit_deleted"]), (1, 1)) + self.assertFalse(proof["backup_deletion_verified"]) + self.assertEqual(len(self.store.project_usage("tenant-a", "project-a")), 1) + self.assertEqual(len(self.store.project_usage("tenant-b", "project-a")), 1) + self.assertEqual(self.call("/v1/models")[0], 200) # Key was not deleted. + def test_stream_is_forwarded_before_generation_finishes(self) -> None: key = self.store.issue_key("tenant-a", "project-a", {"responses:write"}) request = Request(f"http://127.0.0.1:{self.gateway.server_port}/v1/responses", @@ -258,6 +305,13 @@ def test_disconnect_before_upstream_headers_cancels_and_releases_capacity(self) break time.sleep(0.05) self.assertEqual(status, 200) + rows = self.store.project_usage("tenant-a", "project-a") + cancelled = [row for row in rows if row["status"] == 499] + self.assertEqual(len(cancelled), 1) + self.assertIsNone(cancelled[0]["system_one_invocations"]) + self.assertIsNone(cancelled[0]["system_two_invocations"]) + self.assertNotIn("wait_headers", json.dumps(rows)) + self.assertEqual(len({row["request_id"] for row in rows}), len(rows)) if __name__ == "__main__": diff --git a/tests/test_clm_adapter.py b/tests/test_clm_adapter.py index 0bba50a..88c26f6 100644 --- a/tests/test_clm_adapter.py +++ b/tests/test_clm_adapter.py @@ -1,19 +1,20 @@ import math import time import unittest +from collections.abc import Mapping +from typing import cast from c3r.feature_flags import FeatureFlags from c3r.system_one.calibration import CalibrationKey, TemperatureCalibrator -from c3r.system_one.clm_adapter import ClmAdapter, UPSTREAM_CLM_COMMIT +from c3r.system_one.clm_adapter import UPSTREAM_CLM_COMMIT, ClmAdapter from c3r.system_one.fast_path import CalibratedFastPath from c3r.system_one.question_registry import TypedQuestion from tests.test_laya_fast_path import compiled_state - REVISION = "a" * 64 -def ranked(payload: dict[str, object]) -> dict[str, object]: +def ranked(payload: Mapping[str, object]) -> dict[str, object]: options = payload["answers"] assert isinstance(options, list) return { @@ -21,7 +22,7 @@ def ranked(payload: dict[str, object]) -> dict[str, object]: "ranked": [ {"rank": index + 1, "candidate": option, "prob": probability} for index, (option, probability) in enumerate( - zip(reversed(options), (0.9, 0.1), strict=True) + zip(reversed(cast(list[object], options)), (0.9, 0.1), strict=True) ) ] } @@ -53,7 +54,7 @@ def unreachable(_payload: object) -> dict[str, object]: def test_maps_ranked_probabilities_back_to_fixed_option_order(self) -> None: calls: list[object] = [] - def transport(payload: object) -> dict[str, object]: + def transport(payload: Mapping[str, object]) -> dict[str, object]: calls.append(payload) assert isinstance(payload, dict) return ranked(payload) @@ -75,12 +76,13 @@ def test_fails_closed_on_unknown_duplicate_and_nonfinite_candidates(self) -> Non (("NO", 0.1), ("YES", 0.1)), ) for rows in invalid_rows: - response = { + response: dict[str, object] = { "model": "clm-latest", "ranked": [{"candidate": candidate, "prob": prob} for candidate, prob in rows], } with self.subTest(response=response), self.assertRaises(ValueError): - ClmAdapter(revision=REVISION, transport=lambda _payload: response).predict( + ClmAdapter(revision=REVISION, + transport=lambda _payload, response=response: response).predict( compiled_state(), (TypedQuestion("STOP_NOW", ("NO", "YES")),) ) diff --git a/tests/test_cvoc.py b/tests/test_cvoc.py index 82b3e62..445b87b 100644 --- a/tests/test_cvoc.py +++ b/tests/test_cvoc.py @@ -19,6 +19,7 @@ def test_selects_highest_positive_conservative_lower_bound(self) -> None: (local, frontier), estimates ) + assert decision.selected is not None self.assertEqual(decision.selected.id, "frontier") self.assertAlmostEqual(decision.lower_bound, 0.5) diff --git a/tests/test_decisionmix.py b/tests/test_decisionmix.py index 73b58c6..a56d93b 100644 --- a/tests/test_decisionmix.py +++ b/tests/test_decisionmix.py @@ -1,8 +1,8 @@ import json -from dataclasses import replace -from pathlib import Path import tempfile import unittest +from dataclasses import replace +from pathlib import Path from c3r.decisionmix import ( DecisionMixBuilder, diff --git a/tests/test_governed_trace_store.py b/tests/test_governed_trace_store.py index 8b956ba..cd4acd1 100644 --- a/tests/test_governed_trace_store.py +++ b/tests/test_governed_trace_store.py @@ -2,20 +2,22 @@ import sqlite3 import tempfile import unittest -from datetime import datetime, timedelta, timezone +from collections.abc import Mapping +from datetime import UTC, datetime, timedelta from pathlib import Path from c3r.telemetry.governed_store import BoundGovernedTraceSink, GovernedTraceStore, SourceGrant from c3r.telemetry.ledger_anchor import sign_head, verify_anchor from c3r.telemetry.trace import DecisionTrace +NOW = datetime(2026, 9, 22, 12, tzinfo=UTC) -NOW = datetime(2026, 9, 22, 12, tzinfo=timezone.utc) - -def trace(**changes): - fields = dict( - run_id="run_001", +def trace(*, run_id: str = "run_001", + task_outcome: Mapping[str, str | float | bool] | None = None, + artifact_refs: tuple[str, ...] = ()) -> DecisionTrace: + return DecisionTrace( + run_id=run_id, state_hash="a" * 64, access_level="internal", model_provider="deepseek_v4_1_flash", @@ -25,15 +27,12 @@ def trace(**changes): selected_action_id="recommend", authority_result="verified", system_cost={"latency_ms": 18.0}, - task_outcome={"status": "controlled_success"}, - artifact_refs=(), + task_outcome={"status": "controlled_success"} if task_outcome is None else task_outcome, + artifact_refs=artifact_refs, ) - fields.update(changes) - return DecisionTrace(**fields) - class GovernedTraceStoreTests(unittest.TestCase): - def setUp(self): + def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.path = Path(self.temp.name) / "governed.sqlite3" self.grant = SourceGrant( @@ -43,13 +42,13 @@ def setUp(self): rights_attested=True, ) - def tearDown(self): + def tearDown(self) -> None: self.temp.cleanup() - def store(self, *, now=NOW): + def store(self, *, now: datetime = NOW) -> GovernedTraceStore: return GovernedTraceStore(self.path, grants=(self.grant,), clock=lambda: now) - def test_admits_only_attested_registered_internal_task(self): + def test_admits_only_attested_registered_internal_task(self) -> None: with self.store() as store: record = store.append(trace(), source_id="c3r_internal_001", task_id="task_001") self.assertEqual(len(store.records()), 1) @@ -61,12 +60,12 @@ def test_admits_only_attested_registered_internal_task(self): with self.assertRaises(sqlite3.IntegrityError): store.append(trace(), source_id="c3r_internal_001", task_id="task_001") - def test_unattested_source_cannot_be_registered(self): + def test_unattested_source_cannot_be_registered(self) -> None: with self.assertRaisesRegex(ValueError, "rights"): SourceGrant(source_id="imported", owner="wilkont", task_ids=frozenset({"task_001"}), rights_attested=False) - def test_trusted_host_can_bind_source_and_task_for_runtime_sink(self): + def test_trusted_host_can_bind_source_and_task_for_runtime_sink(self) -> None: with self.store() as store: sink = BoundGovernedTraceSink(store, source_id="c3r_internal_001", task_id="task_001") record = sink.append(trace()) @@ -74,7 +73,7 @@ def test_trusted_host_can_bind_source_and_task_for_runtime_sink(self): with self.assertRaisesRegex(ValueError, "unapproved source or task"): BoundGovernedTraceSink(store, source_id="imported", task_id="task_001") - def test_rejects_free_text_or_private_artifact_reference(self): + def test_rejects_free_text_or_private_artifact_reference(self) -> None: with self.store() as store: with self.assertRaisesRegex(ValueError, "redaction"): store.append(trace(task_outcome={"status": "email me at a@example.com"}), @@ -84,7 +83,7 @@ def test_rejects_free_text_or_private_artifact_reference(self): source_id="c3r_internal_001", task_id="task_001") self.assertEqual(store.records(), ()) - def test_purges_after_30_days_and_preserves_remaining_chain(self): + def test_purges_after_30_days_and_preserves_remaining_chain(self) -> None: with self.store(now=NOW - timedelta(days=31)) as store: first = store.append(trace(), source_id="c3r_internal_001", task_id="task_001") with self.store(now=NOW - timedelta(days=1)) as store: @@ -99,7 +98,7 @@ def test_purges_after_30_days_and_preserves_remaining_chain(self): self.assertEqual(remaining[0].previous_hash, first.record_hash) self.assertTrue(store.verify()) - def test_rejects_tampered_database_on_reopen(self): + def test_rejects_tampered_database_on_reopen(self) -> None: with self.store() as store: store.append(trace(), source_id="c3r_internal_001", task_id="task_001") db = sqlite3.connect(self.path) @@ -111,17 +110,17 @@ def test_rejects_tampered_database_on_reopen(self): with self.assertRaisesRegex(ValueError, "hash chain"): self.store() - def test_clock_rollback_cannot_relabel_traces_after_purge(self): + def test_clock_rollback_cannot_relabel_traces_after_purge(self) -> None: with self.store() as store: store.append(trace(), source_id="c3r_internal_001", task_id="task_001") with self.store(now=NOW + timedelta(days=31)) as store: self.assertEqual(store.purge_expired(), 1) - with self.store(now=NOW - timedelta(days=1)) as store: - with self.assertRaisesRegex(ValueError, "clock moved backwards"): - store.append(trace(run_id="run_002"), source_id="c3r_internal_001", - task_id="task_001") + with (self.store(now=NOW - timedelta(days=1)) as store, + self.assertRaisesRegex(ValueError, "clock moved backwards")): + store.append(trace(run_id="run_002"), source_id="c3r_internal_001", + task_id="task_001") - def test_overdue_purge_blocks_new_collection_until_purged(self): + def test_overdue_purge_blocks_new_collection_until_purged(self) -> None: with self.store(now=NOW - timedelta(days=31)) as store: store.append(trace(), source_id="c3r_internal_001", task_id="task_001") with self.store(now=NOW) as store: @@ -134,13 +133,14 @@ def test_overdue_purge_blocks_new_collection_until_purged(self): task_id="task_001") self.assertEqual(len(store.records()), 1) - def test_external_anchor_detects_clean_chain_tail_removal(self): + def test_external_anchor_detects_clean_chain_tail_removal(self) -> None: # The test key stands in for a separate signer; it is not deployment evidence. key = b"fixture-only-signer" - sign = lambda payload: hmac.digest(key, payload, "sha256") - verify = lambda _key_id, payload, signature: hmac.compare_digest( - sign(payload), signature - ) + def sign(payload: bytes) -> bytes: + return hmac.digest(key, payload, "sha256") + + def verify(_key_id: str, payload: bytes, signature: bytes) -> bool: + return hmac.compare_digest(sign(payload), signature) with self.store() as store: store.append(trace(), source_id="c3r_internal_001", task_id="task_001") store.append(trace(run_id="run_002"), source_id="c3r_internal_001", @@ -166,7 +166,7 @@ def test_external_anchor_detects_clean_chain_tail_removal(self): record_hash=current.record_hash, verifier=verify)) - def test_head_snapshot_preserves_purged_prefix_checkpoint(self): + def test_head_snapshot_preserves_purged_prefix_checkpoint(self) -> None: with self.store(now=NOW - timedelta(days=31)) as store: first = store.append(trace(), source_id="c3r_internal_001", task_id="task_001") before = store.snapshot_head(policy_version="fixture-v1") diff --git a/tests/test_host_factory.py b/tests/test_host_factory.py index c70ba06..2aecfe3 100644 --- a/tests/test_host_factory.py +++ b/tests/test_host_factory.py @@ -2,11 +2,15 @@ from c3r.host_factory import ReadOnlyRequestFactory from c3r.state_schema import ( - ActionDefinition, ActionFamily, AuthorityPolicy, RiskClass, ValueEstimate, + ActionDefinition, + ActionFamily, + AuthorityPolicy, + RiskClass, + ValueEstimate, ) -def factory(risk=RiskClass.READ_ONLY): +def factory(risk: RiskClass = RiskClass.READ_ONLY) -> ReadOnlyRequestFactory: return ReadOnlyRequestFactory( definitions=(ActionDefinition( "inspect", ActionFamily.RETRIEVAL, "local", "search", risk, diff --git a/tests/test_http_service.py b/tests/test_http_service.py index 5a5d89d..aba846f 100644 --- a/tests/test_http_service.py +++ b/tests/test_http_service.py @@ -1,24 +1,48 @@ import json import threading import unittest +from collections.abc import Mapping from urllib.error import HTTPError from urllib.request import Request, urlopen +from c3r.candidate_compiler import CandidateCompiler +from c3r.cvoc import RobustCvocController +from c3r.feature_flags import FeatureFlags from c3r.http_service import C3RHTTPServer +from c3r.runtime import RuntimeRequest, StandaloneController +from c3r.state_compiler import StateCompiler +from c3r.telemetry.trace_ledger import TraceLedger +from c3r.verifier_firewall import VerifierFirewall, VerifierPolicy from tests.test_runtime import controller, request - TOKEN = "test-token-with-at-least-thirty-two-characters" class HostFactory: - def build(self, payload): + def build(self, payload: Mapping[str, object]) -> RuntimeRequest: if payload.get("goal") != "Find record": raise ValueError("unknown goal") # Client fields such as policy, estimates, approval and verifier are ignored. return request() +class EffectCapableRuntime(StandaloneController): + """A normally initialized host advertising a forbidden capability.""" + + def __init__(self) -> None: + super().__init__( + flags=FeatureFlags(), compiler=StateCompiler(), candidates=CandidateCompiler(), + cvoc=RobustCvocController(), ledger=TraceLedger(), + verifier=VerifierFirewall( + {}, VerifierPolicy(default_verifier="none"), attestation_key=b"fixture-key", + ), + ) + + @property + def effect_execution_enabled(self) -> bool: + return True + + class HTTPServiceTests(unittest.TestCase): def setUp(self) -> None: runtime, _ = controller() @@ -38,7 +62,7 @@ def tearDown(self) -> None: self.server.server_close() self.thread.join(timeout=2) - def post(self, payload, *, token=TOKEN): + def post(self, payload: Mapping[str, object], *, token: str = TOKEN): body = json.dumps(payload).encode() req = Request( self.base + "/v1/decisions", @@ -89,9 +113,6 @@ def test_non_loopback_bind_is_rejected(self) -> None: ) def test_effect_enabled_runtime_is_rejected(self) -> None: - class EffectCapableRuntime: - effect_execution_enabled = True - with self.assertRaisesRegex(ValueError, "external effects"): C3RHTTPServer( runtime=EffectCapableRuntime(), diff --git a/tests/test_ingress_proxy.py b/tests/test_ingress_proxy.py index e002de8..6855816 100644 --- a/tests/test_ingress_proxy.py +++ b/tests/test_ingress_proxy.py @@ -1,6 +1,7 @@ import json import socket import threading +import time import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.error import HTTPError @@ -12,20 +13,28 @@ UPSTREAM_TOKEN = "upstream-token-that-is-long-enough-for-tests" +class _UpstreamServer(ThreadingHTTPServer): + seen: list[tuple[str, dict[str, str], bytes]] + + class _UpstreamHandler(BaseHTTPRequestHandler): - def log_message(self, _format, *_args): + def log_message(self, format: str, *args: object) -> None: return - def do_GET(self): + def do_GET(self) -> None: + assert isinstance(self.server, _UpstreamServer) self.server.seen.append((self.path, dict(self.headers), b"")) self._reply(200, {"status": "ok"}) - def do_POST(self): - length = int(self.headers["Content-Length"]) + def do_POST(self) -> None: + assert isinstance(self.server, _UpstreamServer) + length_header = self.headers["Content-Length"] + assert length_header is not None + length = int(length_header) self.server.seen.append((self.path, dict(self.headers), self.rfile.read(length))) self._reply(200, {"route": "recommendation"}) - def _reply(self, status, payload): + def _reply(self, status: int, payload: dict[str, str]) -> None: body = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") @@ -35,8 +44,8 @@ def _reply(self, status, payload): class IngressProxyTests(unittest.TestCase): - def setUp(self): - self.upstream = ThreadingHTTPServer(("127.0.0.1", 0), _UpstreamHandler) + def setUp(self) -> None: + self.upstream = _UpstreamServer(("127.0.0.1", 0), _UpstreamHandler) self.upstream.seen = [] self.upstream_thread = threading.Thread(target=self.upstream.serve_forever, daemon=True) self.upstream_thread.start() @@ -52,7 +61,7 @@ def setUp(self): self.ingress_thread.start() self.base = f"http://127.0.0.1:{self.ingress.server_port}" - def tearDown(self): + def tearDown(self) -> None: self.ingress.shutdown() self.ingress.server_close() self.ingress_thread.join(timeout=2) @@ -60,7 +69,8 @@ def tearDown(self): self.upstream.server_close() self.upstream_thread.join(timeout=2) - def request(self, path, *, method="GET", token=CLIENT_TOKEN, payload=None): + def request(self, path: str, *, method: str = "GET", token: str | None = CLIENT_TOKEN, + payload: dict[str, str] | None = None) -> tuple[int, dict[str, object]]: headers = {"Authorization": "Bearer cloud-run-identity-token"} if token is not None: headers["X-C3R-Token"] = token @@ -74,7 +84,7 @@ def request(self, path, *, method="GET", token=CLIENT_TOKEN, payload=None): except HTTPError as error: return error.code, json.load(error) - def test_private_decision_forwards_only_internal_authorization(self): + def test_private_decision_forwards_only_internal_authorization(self) -> None: status, body = self.request("/v1/decisions", method="POST", payload={"goal": "inspect"}) self.assertEqual((status, body["route"]), (200, "recommendation")) path, headers, forwarded = self.upstream.seen[-1] @@ -83,7 +93,42 @@ def test_private_decision_forwards_only_internal_authorization(self): self.assertNotIn("X-C3R-Token", headers) self.assertEqual(json.loads(forwarded), {"goal": "inspect"}) - def test_sdk_bearer_authentication_without_cloud_run_header(self): + def test_idle_connection_cannot_allocate_another_worker_and_release_recovers(self) -> None: + server = C3RIngressServer(upstream_port=self.upstream.server_port, + client_token=CLIENT_TOKEN, upstream_token=UPSTREAM_TOKEN, + host="127.0.0.1", port=0, max_connections=1) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + first = socket.create_connection(("127.0.0.1", server.server_port), timeout=1) + try: + first.sendall(b"GET /health HTTP/1.1\r\n") # Incomplete headers occupy one worker. + time.sleep(0.1) + with socket.create_connection(("127.0.0.1", server.server_port), timeout=1) as second: + second.sendall(b"GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n") + try: + refused = second.recv(4096) + except (ConnectionResetError, ConnectionAbortedError): + refused = b"" + self.assertEqual(refused, b"") + self.assertEqual(self.upstream.seen, []) + first.close() + deadline = time.monotonic() + 2 + while True: + try: + with urlopen(f"http://127.0.0.1:{server.server_port}/health", timeout=1) as reply: + self.assertEqual(reply.status, 200) + break + except OSError: + if time.monotonic() >= deadline: + raise + time.sleep(0.02) + finally: + first.close() + server.shutdown() + server.server_close() + thread.join(timeout=2) + + def test_sdk_bearer_authentication_without_cloud_run_header(self) -> None: request = Request(self.base + "/v1/models", headers={ "Authorization": "Bearer " + CLIENT_TOKEN}) with urlopen(request, timeout=2) as response: @@ -91,7 +136,7 @@ def test_sdk_bearer_authentication_without_cloud_run_header(self): self.assertEqual(self.upstream.seen[-1][1]["Authorization"], "Bearer " + UPSTREAM_TOKEN) - def test_caller_tenant_and_project_headers_are_rejected(self): + def test_caller_tenant_and_project_headers_are_rejected(self) -> None: for name in ("X-Tenant-ID", "X-Project-ID", "OpenAI-Organization", "OpenAI-Project", "X-C3R-Organization", "X-C3R-Project"): with self.subTest(header=name): @@ -102,33 +147,33 @@ def test_caller_tenant_and_project_headers_are_rejected(self): self.assertEqual(raised.exception.code, 401) self.assertEqual(self.upstream.seen, []) - def test_missing_or_wrong_client_token_never_reaches_upstream(self): + def test_missing_or_wrong_client_token_never_reaches_upstream(self) -> None: for token in (None, "wrong", "invalid-café"): status, body = self.request("/v1/decisions", method="POST", token=token, payload={"goal": "inspect"}) self.assertEqual((status, body["error"]), (401, "unauthorized")) self.assertEqual(self.upstream.seen, []) - def test_health_is_unprivileged_but_metrics_require_token(self): + def test_health_is_unprivileged_but_metrics_require_token(self) -> None: self.assertEqual(self.request("/health", token=None)[0], 200) self.assertEqual(self.request("/metrics", token=None)[0], 401) self.assertEqual(self.request("/ready", token=None)[0], 401) self.assertEqual(self.request("/v1/models", token=None)[0], 401) - def test_stateless_paths_forward_without_client_credential_leak(self): + def test_stateless_paths_forward_without_client_credential_leak(self) -> None: for path in ("/v1/c3r/decide", "/v1/c3r/rank", "/v1/system-one"): self.assertEqual(self.request(path, method="POST", payload={"goal": "inspect"})[0], 200) seen_path, headers, _ = self.upstream.seen[-1] self.assertEqual(seen_path, path) self.assertNotIn("X-C3R-Token", headers) - def test_rejects_unknown_path_without_contacting_upstream(self): + def test_rejects_unknown_path_without_contacting_upstream(self) -> None: self.assertEqual(self.request("/admin")[0], 404) self.assertEqual(self.request("/v1/decisions?debug=1", method="POST", payload={"goal": "inspect"})[0], 404) self.assertEqual(self.upstream.seen, []) - def test_rejects_ambiguous_framing_and_non_json_body(self): + def test_rejects_ambiguous_framing_and_non_json_body(self) -> None: request = Request( self.base + "/v1/decisions", data=b"{}", @@ -155,13 +200,13 @@ def test_rejects_ambiguous_framing_and_non_json_body(self): self.assertTrue(response.startswith((b"HTTP/1.0 400 ", b"HTTP/1.0 413 "))) self.assertEqual(self.upstream.seen, []) - def test_upstream_failure_is_not_mistaken_for_success(self): + def test_upstream_failure_is_not_mistaken_for_success(self) -> None: self.upstream.shutdown() self.upstream.server_close() status, body = self.request("/v1/decisions", method="POST", payload={"goal": "inspect"}) self.assertEqual((status, body["error"]), (503, "upstream_unavailable")) - def test_upstream_route_must_be_loopback_and_secrets_distinct(self): + def test_upstream_route_must_be_loopback_and_secrets_distinct(self) -> None: with self.assertRaisesRegex(ValueError, "loopback"): C3RIngressServer(upstream_host="example.com", upstream_port=8081, client_token=CLIENT_TOKEN, upstream_token=UPSTREAM_TOKEN, diff --git a/tests/test_key_recovery_cli.py b/tests/test_key_recovery_cli.py new file mode 100644 index 0000000..b424d83 --- /dev/null +++ b/tests/test_key_recovery_cli.py @@ -0,0 +1,163 @@ +"""Credential recovery through the operator-local CLI; no live cloud state.""" +import json +import os +import sqlite3 +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from c3r.api_access import AccessStore + + +class KeyRecoveryCLITests(unittest.TestCase): + def run_cli(self, database: Path, command: str, destination: Path, + *extra: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, "-m", "c3r.key_management", "--database", str(database), + command, "--destination", str(destination), *extra], + capture_output=True, text=True, timeout=5, check=False) + + def test_backup_and_restore_preserve_projects_but_do_not_reactivate_old_keys(self) -> None: + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + source = AccessStore(root / "source.sqlite3") + source.create_project("avori", "staging") + key = source.issue_key("avori", "staging", {"models:read"}) + backup = root / "backup.sqlite3" + restored = root / "restored.sqlite3" + backup_hash = "" + for command, database, destination in ( + ("backup", source.path, backup), ("restore", backup, restored)): + arguments = [sys.executable, "-m", "c3r.key_management", "--database", str(database), + command, "--destination", str(destination)] + if command == "restore": + arguments += ["--expected-sha256", backup_hash] + result = subprocess.run(arguments, + capture_output=True, text=True, timeout=5, check=False) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + receipt = json.loads(result.stdout) + self.assertEqual(receipt["keys_copied"], 1) + self.assertFalse(receipt["encryption_verified"]) + self.assertNotIn(key.secret, result.stdout + result.stderr) + if command == "backup": + backup_hash = receipt["sha256"] + recovered = AccessStore(restored) + self.assertIsNone(recovered.authenticate(key.secret)) + self.assertIsNotNone(source.authenticate(key.secret)) + self.assertEqual(recovered.list_keys("avori", "staging")[0]["revoked"], 1) + replacement = recovered.issue_key("avori", "staging", {"models:read"}) + self.assertIsNotNone(recovered.authenticate(replacement.secret)) + + def test_backup_excludes_usage_history_and_plaintext_keys(self) -> None: + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + source = AccessStore(root / "source.sqlite3") + source.create_project("avori", "staging", rpm=1, key_rps=1) + key = source.issue_key("avori", "staging", {"models:read"}) + principal = source.authenticate(key.secret) + assert principal is not None + source.dispatch_event(principal, "unique-request-history-marker") + source.record_usage(principal, "unique-request-history-marker", route="/v1/models", + model=None, status=200, latency_ms=5, input_tokens=None, + output_tokens=None, system_one_invocations=0, system_two_invocations=0) + backup = root / "backup.sqlite3" + result = self.run_cli(source.path, "backup", backup) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + snapshot = AccessStore(backup) + self.assertEqual(snapshot.project_usage("avori", "staging"), []) + self.assertEqual(len(source.project_usage("avori", "staging")), 1) + content = backup.read_bytes() + self.assertNotIn(b"unique-request-history-marker", content) + self.assertNotIn(key.secret.encode(), content) + snapshot_principal = snapshot.authenticate(key.secret) + assert snapshot_principal is not None + self.assertTrue(snapshot.admit(snapshot_principal)) + self.assertFalse(snapshot.admit(snapshot_principal)) + + def test_backup_cannot_overwrite_source_or_an_existing_snapshot(self) -> None: + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + source = AccessStore(root / "source.sqlite3") + source.create_project("avori", "staging") + key = source.issue_key("avori", "staging", {"models:read"}) + backup = root / "backup.sqlite3" + self.assertEqual(self.run_cli(source.path, "backup", backup).returncode, 0) + prior = backup.read_bytes() + for destination in (backup, source.path): + with self.subTest(destination=destination.name): + result = self.run_cli(source.path, "backup", destination) + self.assertEqual(result.returncode, 1) + self.assertNotIn(key.secret, result.stdout + result.stderr) + self.assertEqual(backup.read_bytes(), prior) + self.assertIsNotNone(source.authenticate(key.secret)) + + def test_restore_hash_mismatch_and_missing_source_create_no_database(self) -> None: + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + source = AccessStore(root / "source.sqlite3") + backup = root / "backup.sqlite3" + self.assertEqual(self.run_cli(source.path, "backup", backup).returncode, 0) + destination = root / "restored.sqlite3" + mismatch = self.run_cli(backup, "restore", destination, "--expected-sha256", "0" * 64) + self.assertEqual(mismatch.returncode, 1) + self.assertFalse(destination.exists()) + missing = root / "missing.sqlite3" + self.assertEqual(self.run_cli(missing, "backup", destination).returncode, 1) + self.assertFalse(missing.exists()) + self.assertFalse(destination.exists()) + + def test_restore_rejects_ordinary_database_even_with_matching_hash(self) -> None: + import hashlib + + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + source = AccessStore(root / "source.sqlite3") + digest = hashlib.sha256(source.path.read_bytes()).hexdigest() + result = self.run_cli(source.path, "restore", root / "restored.sqlite3", + "--expected-sha256", digest) + self.assertEqual(result.returncode, 1) + + def test_restore_rejects_unbound_wal_sidecar(self) -> None: + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + source = AccessStore(root / "source.sqlite3") + source.create_project("avori", "staging") + backup = root / "backup.sqlite3" + self.assertEqual(self.run_cli(source.path, "backup", backup).returncode, 0) + writer = sqlite3.connect(backup) + try: + writer.execute("PRAGMA journal_mode=WAL") + # Capture the main-file hash after WAL mode is selected, before + # committing new contents only into its unbound sidecar. + import hashlib + digest = hashlib.sha256(backup.read_bytes()).hexdigest() + writer.execute("UPDATE projects SET rpm=999") + writer.commit() + self.assertEqual(hashlib.sha256(backup.read_bytes()).hexdigest(), digest) + destination = root / "restored.sqlite3" + result = self.run_cli(backup, "restore", destination, + "--expected-sha256", digest) + self.assertEqual(result.returncode, 1) + self.assertFalse(destination.exists()) + finally: + writer.close() + + def test_backup_rejects_empty_and_hardlinked_sources_without_modification(self) -> None: + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + empty = root / "empty.sqlite3" + empty.touch() + result = self.run_cli(empty, "backup", root / "empty-backup.sqlite3") + self.assertEqual(result.returncode, 1) + self.assertEqual(empty.read_bytes(), b"") + source = AccessStore(root / "source.sqlite3") + alias = root / "alias.sqlite3" + os.link(source.path, alias) + result = self.run_cli(alias, "backup", root / "alias-backup.sqlite3") + self.assertEqual(result.returncode, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_laya_fast_path.py b/tests/test_laya_fast_path.py index b0da96f..9b0b504 100644 --- a/tests/test_laya_fast_path.py +++ b/tests/test_laya_fast_path.py @@ -1,5 +1,6 @@ import unittest from dataclasses import dataclass +from typing import NoReturn from c3r.state_schema import CompiledState from c3r.system_one.calibration import CalibrationKey, TemperatureCalibrator @@ -103,7 +104,7 @@ def test_pinned_backend_verifies_revision_and_license_before_loading(self) -> No @dataclass class FakeAgent: - def predict(self, _state: object, _questions: object) -> dict[str, object]: + def predict(self, state: object, questions: object) -> dict[str, object]: return { "answers": { "STOP_NOW": { @@ -129,6 +130,8 @@ def predict(self, _state: object, _questions: object) -> dict[str, object]: def test_pinned_backend_rejects_upstream_license_mismatch(self) -> None: revision = "1c5edc17a7acd8701df6fc341c0d179f1c62c982" + def unreachable_loader(_path: str, _device: str | None) -> NoReturn: + self.fail("license mismatch reached the loader") backend = PinnedLayaBackend( model_id="convaiinnovations/laya", revision=revision, @@ -136,7 +139,7 @@ def test_pinned_backend_rejects_upstream_license_mismatch(self) -> None: sha=revision, license="unknown" ), snapshot_fetcher=lambda _model, _revision: "C:/model", - agent_loader=lambda _path, _device: object(), + agent_loader=unreachable_loader, ) with self.assertRaisesRegex(ValueError, "license"): @@ -147,7 +150,7 @@ def test_pinned_backend_drops_non_finite_probabilities(self) -> None: @dataclass class FakeAgent: - def predict(self, _state: object, _questions: object) -> dict[str, object]: + def predict(self, state: object, questions: object) -> dict[str, object]: return { "answers": { "STOP_NOW": { diff --git a/tests/test_ledger_anchor.py b/tests/test_ledger_anchor.py index 064f14c..0a8d9f9 100644 --- a/tests/test_ledger_anchor.py +++ b/tests/test_ledger_anchor.py @@ -1,7 +1,7 @@ import hmac import unittest from dataclasses import replace -from datetime import datetime, timezone +from datetime import UTC, datetime from c3r.telemetry.ledger_anchor import capture_head, sign_head, verify_anchor from c3r.telemetry.trace import DecisionTrace @@ -29,10 +29,11 @@ class LedgerAnchorTests(unittest.TestCase): def test_signed_head_matches_chain_and_detects_removal_or_alteration(self) -> None: # Test-only signer. Deployment must use a key unavailable to the writer. test_key = b"fixture-only-key" - sign = lambda payload: hmac.digest(test_key, payload, "sha256") - verify = lambda _key_id, payload, signature: hmac.compare_digest( - sign(payload), signature - ) + def sign(payload: bytes) -> bytes: + return hmac.digest(test_key, payload, "sha256") + + def verify(_key_id: str, payload: bytes, signature: bytes) -> bool: + return hmac.compare_digest(sign(payload), signature) ledger = TraceLedger() ledger.append(trace("one")) ledger.append(trace("two")) @@ -40,7 +41,7 @@ def test_signed_head_matches_chain_and_detects_removal_or_alteration(self) -> No sequence=2, record_hash=ledger.records[-1].record_hash, policy_version="fixture-v1", - clock=lambda: datetime(2026, 9, 23, tzinfo=timezone.utc), + clock=lambda: datetime(2026, 9, 23, tzinfo=UTC), ) anchor = sign_head(head, key_id="test-only", signer=sign) diff --git a/tests/test_provider_bridge.py b/tests/test_provider_bridge.py index 2e2e5b2..738d8c9 100644 --- a/tests/test_provider_bridge.py +++ b/tests/test_provider_bridge.py @@ -1,5 +1,6 @@ import json import unittest +from typing import cast from c3r.adapters.providers import ( ProviderAdapter, @@ -22,9 +23,11 @@ def state(data_boundary: str): class ProviderBridgeTests(unittest.TestCase): def test_local_deepseek_receives_compiled_state_and_returns_plan(self) -> None: - captured = [] + captured: list[dict[str, object]] = [] - def transport(_url, _headers, payload): + def transport( + _url: str, _headers: dict[str, str], payload: dict[str, object], + ) -> TransportResponse: captured.append(payload) return TransportResponse( 200, @@ -52,16 +55,27 @@ def transport(_url, _headers, payload): self.assertEqual(result.deliberation.plan, ("inspect",)) self.assertEqual(result.observed_cost["latency_ms"], 12.0) - self.assertIn("Plan a read-only check", captured[0]["messages"][1]["content"]) + messages = captured[0]["messages"] + assert isinstance(messages, list) + message = cast(list[object], messages)[1] + assert isinstance(message, dict) + content = cast(dict[object, object], message)["content"] + assert isinstance(content, str) + self.assertIn("Plan a read-only check", content) def test_remote_provider_is_not_called_for_local_data(self) -> None: - calls = [] + calls: list[bool] = [] + def record( + _url: str, _headers: dict[str, str], _payload: dict[str, object], + ) -> TransportResponse: + calls.append(True) + return TransportResponse(200, {}, 0) adapter = ProviderAdapter( ProviderConfig( "frontier", ProviderKind.OPENAI_COMPATIBLE, "https://example.invalid/v1", "frontier-model", "test-key", ), - transport=lambda *_: calls.append(True), + transport=record, ) with self.assertRaises(ValueError): diff --git a/tests/test_responses_stream.py b/tests/test_responses_stream.py index 5d17d03..3ba4892 100644 --- a/tests/test_responses_stream.py +++ b/tests/test_responses_stream.py @@ -19,12 +19,26 @@ TOKEN = "stream-test-token-with-at-least-thirty-two-characters" +class _GenerationServer(ThreadingHTTPServer): + requests: list[dict[str, object]] + hold_headers: bool + headers_pending: threading.Event + hold_after_first: bool + fail_after_first: bool + finish_with_text: bool + release_next: threading.Event + abort_seen: threading.Event + + class _GenerationHandler(BaseHTTPRequestHandler): - def log_message(self, *_args): + def log_message(self, format: str, *args: object) -> None: return - def do_POST(self): - body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + def do_POST(self) -> None: + assert isinstance(self.server, _GenerationServer) + length_header = self.headers["Content-Length"] + assert length_header is not None + body = json.loads(self.rfile.read(int(length_header))) self.server.requests.append(body) if self.server.hold_headers: self.server.headers_pending.set() @@ -38,7 +52,7 @@ def do_POST(self): self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.end_headers() - events = [ + events: list[dict[str, object]] = [ {"choices": [{"delta": {"content": "Check ", "reasoning_content": "PRIVATE"}, "finish_reason": None}]}, {"choices": [{"delta": {"content": "charges."}, @@ -74,8 +88,8 @@ def do_POST(self): class ResponsesStreamTests(unittest.TestCase): - def setUp(self): - self.backend = ThreadingHTTPServer(("127.0.0.1", 0), _GenerationHandler) + def setUp(self) -> None: + self.backend = _GenerationServer(("127.0.0.1", 0), _GenerationHandler) self.backend.requests = [] self.backend.hold_headers = False self.backend.headers_pending = threading.Event() @@ -106,7 +120,7 @@ def setUp(self): self.api_thread = threading.Thread(target=self.api.serve_forever, daemon=True) self.api_thread.start() - def tearDown(self): + def tearDown(self) -> None: self.backend.release_next.set() self.api.shutdown() self.api.server_close() @@ -115,7 +129,7 @@ def tearDown(self): self.backend.server_close() self.backend_thread.join(timeout=2) - def test_stream_emits_real_text_deltas_and_terminal_response(self): + def test_stream_emits_real_text_deltas_and_terminal_response(self) -> None: connection = HTTPConnection("127.0.0.1", self.api.server_port, timeout=3) connection.request("POST", "/v1/responses", body=json.dumps({ "model": "c3r-core", "input": "Find record", "stream": True, @@ -133,10 +147,15 @@ def test_stream_emits_real_text_deltas_and_terminal_response(self): self.assertIn("Check charges.", wire) self.assertNotIn("PRIVATE", wire) self.assertTrue(self.backend.requests[0]["stream"]) + terminal = [json.loads(line[6:]) for line in wire.splitlines() + if line.startswith("data: ")][-1]["response"] + self.assertEqual((terminal["c3r"]["system_one_invocations"], + terminal["c3r"]["system_two_invocations"]), (0, 1)) - def test_rejected_stream_does_not_start_backend_generation(self): + def test_rejected_stream_does_not_start_backend_generation(self) -> None: runtime, _ = controller(deliberative=True, accepted=False) self.api.runtime = runtime + assert self.api.responses is not None self.api.responses.runtime = runtime connection = HTTPConnection("127.0.0.1", self.api.server_port, timeout=3) connection.request("POST", "/v1/responses", body=json.dumps({ @@ -148,7 +167,7 @@ def test_rejected_stream_does_not_start_backend_generation(self): connection.close() self.assertEqual(self.backend.requests, []) - def test_backend_error_emits_failed_terminal_event_without_private_reasoning(self): + def test_backend_error_emits_failed_terminal_event_without_private_reasoning(self) -> None: self.backend.fail_after_first = True connection = HTTPConnection("127.0.0.1", self.api.server_port, timeout=3) connection.request("POST", "/v1/responses", body=json.dumps({ @@ -162,7 +181,7 @@ def test_backend_error_emits_failed_terminal_event_without_private_reasoning(sel self.assertNotIn("event: response.completed", wire) self.assertNotIn("PRIVATE", wire) - def test_final_delta_with_finish_reason_is_not_lost(self): + def test_final_delta_with_finish_reason_is_not_lost(self) -> None: self.backend.finish_with_text = True connection = HTTPConnection("127.0.0.1", self.api.server_port, timeout=3) connection.request("POST", "/v1/responses", body=json.dumps({ @@ -175,7 +194,7 @@ def test_final_delta_with_finish_reason_is_not_lost(self): self.assertIn('"text":"Check charges."', wire) self.assertIn("event: response.completed", wire) - def test_disconnect_before_backend_headers_cancels_generation(self): + def test_disconnect_before_backend_headers_cancels_generation(self) -> None: self.api.generation_capacity = threading.BoundedSemaphore(1) self.backend.hold_headers = True body = json.dumps({"model": "c3r-core", "input": "Find record", "stream": True}).encode() @@ -197,10 +216,13 @@ def test_disconnect_before_backend_headers_cancels_generation(self): response.read() second.close() - def test_disconnected_stream_releases_generation_capacity(self): + def test_disconnected_stream_releases_generation_capacity(self) -> None: self.api.generation_capacity = threading.BoundedSemaphore(1) self.backend.hold_after_first = True first = HTTPConnection("127.0.0.1", self.api.server_port, timeout=3) + first.connect() + client_socket = first.sock + assert client_socket is not None body = json.dumps({"model": "c3r-core", "input": "Find record", "stream": True}) headers = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"} first.request("POST", "/v1/responses", body=body, headers=headers) @@ -214,7 +236,7 @@ def test_disconnected_stream_releases_generation_capacity(self): self.assertEqual(refused.status, 429) refused.read() second.close() - first_response.fp.raw._sock.shutdown(socket.SHUT_RDWR) + client_socket.shutdown(socket.SHUT_RDWR) first_response.close() first.close() deadline = time.monotonic() + 1 diff --git a/tests/test_retention_job.py b/tests/test_retention_job.py index 5eb817e..59f7682 100644 --- a/tests/test_retention_job.py +++ b/tests/test_retention_job.py @@ -1,34 +1,39 @@ import io +import json import unittest -from datetime import datetime, timedelta, timezone +from collections.abc import Iterable +from datetime import UTC, datetime, timedelta from unittest.mock import patch +from urllib.request import Request from c3r.retention_job import ( - DELETE_AFTER_DAYS, GcsJsonClient, StoredObject, _created_at, - _verify_runtime_project, bucket_from_environment, purge, + DELETE_AFTER_DAYS, + GcsJsonClient, + StoredObject, + bucket_from_environment, + purge, ) - -NOW = datetime(2026, 9, 23, 16, 0, tzinfo=timezone.utc) +NOW = datetime(2026, 9, 23, 16, 0, tzinfo=UTC) BUCKET = "colomboai-c3r-staging-traces-123456789012" class FakeClient: - def __init__(self, objects): + def __init__(self, objects: Iterable[StoredObject]) -> None: self.bucket = BUCKET self.objects = list(objects) - self.deleted = [] + self.deleted: list[tuple[str, int]] = [] - def list_all(self): + def list_all(self) -> tuple[StoredObject, ...]: return tuple(self.objects) - def delete_generation(self, obj): + def delete_generation(self, obj: StoredObject) -> None: self.deleted.append((obj.name, obj.generation)) self.objects = [item for item in self.objects if item != obj] class RetentionJobTests(unittest.TestCase): - def test_deletes_only_expired_generation_and_verifies_empty_overdue_set(self): + def test_deletes_only_expired_generation_and_verifies_empty_overdue_set(self) -> None: old = StoredObject("traces/old", 3, NOW - timedelta(days=DELETE_AFTER_DAYS)) fresh = StoredObject("traces/new", 4, NOW - timedelta(days=1)) client = FakeClient([old, fresh]) @@ -40,24 +45,24 @@ def test_deletes_only_expired_generation_and_verifies_empty_overdue_set(self): self.assertEqual(report["expired_remaining"], 0) self.assertFalse(report["trace_collection_enabled"]) - def test_surviving_expired_object_fails_closed(self): + def test_surviving_expired_object_fails_closed(self) -> None: old = StoredObject("traces/old", 3, NOW - timedelta(days=29)) class NonDeletingClient(FakeClient): - def delete_generation(self, obj): + def delete_generation(self, obj: StoredObject) -> None: self.deleted.append((obj.name, obj.generation)) with self.assertRaisesRegex(RuntimeError, "expired objects remain"): purge(NonDeletingClient([old]), now=NOW) - def test_duplicate_inventory_and_non_utc_clock_rejected(self): + def test_duplicate_inventory_and_non_utc_clock_rejected(self) -> None: old = StoredObject("traces/old", 3, NOW - timedelta(days=29)) with self.assertRaisesRegex(ValueError, "duplicate"): purge(FakeClient([old, old]), now=NOW) with self.assertRaisesRegex(ValueError, "UTC"): purge(FakeClient([]), now=NOW.replace(tzinfo=None)) - def test_distinct_generations_are_purged_without_treating_them_as_duplicates(self): + def test_distinct_generations_are_purged_without_treating_them_as_duplicates(self) -> None: old = StoredObject("traces/replaced", 3, NOW - timedelta(days=29)) fresh = StoredObject("traces/replaced", 4, NOW - timedelta(days=1)) client = FakeClient([old, fresh]) @@ -68,36 +73,50 @@ def test_distinct_generations_are_purged_without_treating_them_as_duplicates(sel self.assertEqual(client.objects, [fresh]) self.assertEqual(report["expired_remaining"], 0) - def test_gcs_transport_lists_versions_and_deletes_exact_generation(self): - client = object.__new__(GcsJsonClient) - client.bucket = BUCKET - client._storage_api = "https://storage.googleapis.com/storage/v1/b/" + BUCKET + "/o" - requests = [] - - def fake_request(method, url): - requests.append((method, url)) - if method == "GET": - return {"items": [ + def test_gcs_transport_lists_versions_and_deletes_exact_generation(self) -> None: + requests: list[tuple[str, str]] = [] + + def fake_open(request: Request, *, timeout: int) -> io.BytesIO: + requests.append((request.get_method(), request.full_url)) + if "numeric-project-id" in request.full_url: + return io.BytesIO(b"123456789012") + if "/token" in request.full_url: + return io.BytesIO(b'{"access_token":"' + b"x" * 24 + b'"}') + if request.get_method() == "GET": + return io.BytesIO(json.dumps({"items": [ {"name": "traces/replaced", "generation": "3", "timeCreated": "2026-08-01T00:00:00Z"}, {"name": "traces/replaced", "generation": "4", "timeCreated": "2026-09-23T00:00:00Z"}, - ]} - return None + ]}).encode()) + return io.BytesIO(b"") - client._request = fake_request - objects = client.list_all() - client.delete_generation(objects[0]) + with patch("c3r.retention_job.urlopen", side_effect=fake_open): + client = GcsJsonClient(BUCKET) + objects = client.list_all() + client.delete_generation(objects[0]) + requests = requests[2:] self.assertEqual([obj.generation for obj in objects], [3, 4]) self.assertIn("versions=true", requests[0][1]) self.assertIn("generation=3", requests[1][1]) self.assertNotIn("ifGenerationMatch", requests[1][1]) - def test_creation_timestamp_must_be_timezone_aware(self): - self.assertEqual(_created_at("2026-09-23T16:00:00Z"), NOW) - with self.assertRaisesRegex(ValueError, "UTC offset"): - _created_at("2026-09-23T16:00:00") - - def test_dedicated_bucket_must_be_explicit_and_narrow(self): + def test_creation_timestamp_must_be_timezone_aware(self) -> None: + for timestamp in ("2026-09-23T16:00:00Z", "2026-09-23T16:00:00"): + inventory = json.dumps({"items": [{"name": "traces/time", "generation": "1", + "timeCreated": timestamp}]}).encode() + with patch("c3r.retention_job.urlopen", side_effect=[ + io.BytesIO(b"123456789012"), + io.BytesIO(b'{"access_token":"' + b"x" * 24 + b'"}'), + io.BytesIO(inventory), + ]): + client = GcsJsonClient(BUCKET) + if timestamp.endswith("Z"): + self.assertEqual(client.list_all()[0].created_at, NOW) + else: + with self.assertRaisesRegex(ValueError, "UTC offset"): + client.list_all() + + def test_dedicated_bucket_must_be_explicit_and_narrow(self) -> None: self.assertEqual(bucket_from_environment({"C3R_TRACE_BUCKET": BUCKET}), BUCKET) for value in ("", "colomboai-c3r-private-traces-123456789012", "unrelated-bucket", "colomboai-c3r-staging-traces-123456789012/other"): @@ -108,18 +127,23 @@ def test_dedicated_bucket_must_be_explicit_and_narrow(self): with self.assertRaisesRegex(ValueError, "C3R_TRACE_BUCKET"): purge(client, now=NOW) - def test_runtime_project_must_match_bucket_suffix(self): - _verify_runtime_project(BUCKET, "123456789012") - with self.assertRaisesRegex(ValueError, "project number"): - _verify_runtime_project(BUCKET, "999999999999") - - def test_gcs_client_rejects_wrong_project_before_credential_request(self): - with patch("c3r.retention_job.urlopen", return_value=io.BytesIO(b"999999999999")) as open_url: - with self.assertRaisesRegex(ValueError, "project number"): - GcsJsonClient(BUCKET) + def test_runtime_project_must_match_bucket_suffix(self) -> None: + with patch("c3r.retention_job.urlopen", side_effect=[ + io.BytesIO(b"123456789012"), + io.BytesIO(b'{"access_token":"' + b"x" * 24 + b'"}'), + ]): + self.assertEqual(GcsJsonClient(BUCKET).bucket, BUCKET) + with (patch("c3r.retention_job.urlopen", return_value=io.BytesIO(b"999999999999")), + self.assertRaisesRegex(ValueError, "project number")): + GcsJsonClient(BUCKET) + + def test_gcs_client_rejects_wrong_project_before_credential_request(self) -> None: + with (patch("c3r.retention_job.urlopen", return_value=io.BytesIO(b"999999999999")) as open_url, + self.assertRaisesRegex(ValueError, "project number")): + GcsJsonClient(BUCKET) self.assertEqual(open_url.call_count, 1) - def test_gcs_client_accepts_metadata_bound_bucket(self): + def test_gcs_client_accepts_metadata_bound_bucket(self) -> None: token = b'{"access_token":"' + b"x" * 24 + b'"}' with patch("c3r.retention_job.urlopen", side_effect=[ io.BytesIO(b"123456789012"), io.BytesIO(token), @@ -131,4 +155,3 @@ def test_gcs_client_accepts_metadata_bound_bucket(self): if __name__ == "__main__": unittest.main() - diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 4c2f486..026ad75 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -1,22 +1,28 @@ import unittest +from collections.abc import Callable, Mapping from dataclasses import replace +from typing import cast from c3r.adapters.providers import ProviderExecutionResult from c3r.candidate_compiler import CandidateCompiler from c3r.cvoc import RobustCvocController from c3r.deliberative.envelope import DeliberativeResult from c3r.feature_flags import FeatureFlags +from c3r.runtime import Deliberator as RuntimeDeliberator from c3r.runtime import RuntimeRequest, StandaloneController from c3r.state_compiler import StateCompiler from c3r.state_schema import ( + ActionCandidate, ActionDefinition, ActionFamily, AuthorityPolicy, + CompiledState, Provenance, RawState, RiskClass, ValueEstimate, ) +from c3r.system_one.advisory import AdvisoryFastPath from c3r.system_one.calibration import TemperatureCalibrator from c3r.system_one.clm_adapter import ClmAdapter from c3r.system_one.fast_path import CalibratedFastPath, LayaFastPath @@ -24,7 +30,6 @@ from c3r.telemetry.trace_ledger import TraceLedger from c3r.verifier_firewall import VerifierDecision, VerifierFirewall, VerifierPolicy - KEY = b"verification-test-key" ACTION_ID = "lookup:0:local:policy" @@ -71,9 +76,9 @@ def controller( system_one: bool = False, deliberative: bool = False, accepted: bool = True, - executor=None, - fast_path=None, - deliberator=None, + executor: Callable[[ActionCandidate], None] | None = None, + fast_path: CalibratedFastPath | AdvisoryFastPath | None = None, + deliberator: RuntimeDeliberator | None = None, system_one_provider: str = "clm", ) -> tuple[StandaloneController, TraceLedger]: ledger = TraceLedger() @@ -103,7 +108,7 @@ def controller( class RuntimeTests(unittest.TestCase): def test_executor_configuration_is_rejected_before_any_effect(self) -> None: - effects = [] + effects: list[ActionCandidate] = [] with self.assertRaisesRegex(ValueError, "external effects"): controller(executor=effects.append) @@ -172,7 +177,7 @@ def test_uncalibrated_system_one_abstains_into_non_authoritative_deliberation(se fast_path = LayaFastPath(adapter=adapter, calibrator=TemperatureCalibrator({})) class Deliberator: - def deliberate(self, _state): + def deliberate(self, state: CompiledState) -> object: return {"plan": ["inspect"]} runtime, _ = controller( @@ -203,8 +208,11 @@ def test_default_clm_never_silently_runs_laya(self) -> None: self.assertEqual(outcome.reason, "SYSTEM_ONE_PROVIDER_MISMATCH") def test_clm_rank_is_advisory_and_abstains_without_calibration(self) -> None: - def rank(payload): + def rank(payload: Mapping[str, object]) -> Mapping[str, object]: options = payload["answers"] + if not isinstance(options, list): + raise TypeError("fixture answers must be a list") + options = tuple(str(option) for option in cast(list[object], options)) probability = 1.0 / len(options) return { "model": "clm-latest", @@ -223,15 +231,16 @@ def rank(payload): ) outcome = runtime.run(request()) self.assertEqual(outcome.reason, "SYSTEM_ONE_ABSTAINED_NO_PROVIDER") + assert outcome.fast_path is not None self.assertEqual(outcome.fast_path.candidate_probabilities, (1.0,)) self.assertIn('"model_provider":"Contrastive-LM/CLM"', ledger.records[-1].canonical_json) def test_clm_outage_escalates_without_granting_authority(self) -> None: - def unavailable(_payload): + def unavailable(_payload: Mapping[str, object]) -> Mapping[str, object]: raise OSError("CLM unavailable") class Deliberator: - def deliberate(self, _state): + def deliberate(self, state: CompiledState) -> object: return {"plan": ["inspect"]} runtime, ledger = controller( @@ -251,7 +260,7 @@ def deliberate(self, _state): def test_provider_usage_is_recorded_without_granting_authority(self) -> None: class Deliberator: - def deliberate(self, _state): + def deliberate(self, state: CompiledState) -> ProviderExecutionResult: return ProviderExecutionResult( DeliberativeResult(("inspect",), (), (), (), ()), {"latency_ms": 12.0, "input_tokens": 10.0}, @@ -284,7 +293,7 @@ def deliberate(self, _state): def test_model_requested_unsafe_action_never_reaches_executor(self) -> None: class Deliberator: - def deliberate(self, _state): + def deliberate(self, state: CompiledState) -> ProviderExecutionResult: return ProviderExecutionResult( DeliberativeResult((), (), (), (), ("delete all records",)), {"latency_ms": 1.0}, "untrusted-model", "fixture", @@ -308,11 +317,12 @@ def deliberate(self, _state): self.assertEqual(outcome.route, "deliberative") self.assertFalse(runtime.effect_execution_enabled) + assert isinstance(outcome.deliberation, DeliberativeResult) self.assertEqual(outcome.deliberation.requested_actions, ("delete all records",)) def test_provider_outage_falls_back_without_effect(self) -> None: class Deliberator: - def deliberate(self, _state): + def deliberate(self, state: CompiledState) -> object: raise OSError("provider unavailable") runtime, _ = controller(deliberative=True, deliberator=Deliberator()) diff --git a/tests/test_serve.py b/tests/test_serve.py index 6a9c280..a40dd36 100644 --- a/tests/test_serve.py +++ b/tests/test_serve.py @@ -19,7 +19,7 @@ from c3r.state_compiler import StateCompiler from c3r.telemetry.ephemeral import EphemeralTraceSink from c3r.verifier_firewall import VerifierFirewall, VerifierPolicy -from tests.test_http_service import HostFactory +from tests.test_http_service import EffectCapableRuntime, HostFactory from tests.test_runtime import controller CLIENT_TOKEN = "client-token-with-at-least-thirty-two-characters" @@ -49,7 +49,7 @@ def test_internal_token_without_explicit_internal_port_fails_startup(self): "C3R_INTERNAL_READY_TOKEN": "internal-test-token-never-an-api-token"}) result = subprocess.run([sys.executable, "-m", "c3r.serve"], env={**os.environ, **values}, capture_output=True, - text=True, timeout=3) + text=True, timeout=3, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("C3R_INTERNAL_READY_PORT", result.stderr) @@ -106,7 +106,7 @@ def test_production_entrypoint_requires_key_mode_database_before_binding(self): "C3R_DELIBERATIVE": "true", "C3R_CLM_CONTAINER_DIGEST": "sha256:" + "a" * 64}) result = subprocess.run([sys.executable, "-m", "c3r.serve"], env={**os.environ, **values}, capture_output=True, - text=True, timeout=5) + text=True, timeout=5, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("C3R_API_ACCESS_DB", result.stderr) @@ -131,9 +131,6 @@ def test_invalid_port_and_equal_tokens_fail(self): build_servers(values, builder_loader=lambda _: lambda: (controller()[0], HostFactory())) def test_effect_enabled_host_is_rejected(self): - class EffectCapableRuntime: - effect_execution_enabled = True - with self.assertRaisesRegex(ValueError, "recommendation-only"): build_servers( config(), diff --git a/tests/test_sqlite_ledger.py b/tests/test_sqlite_ledger.py index bb98661..4214702 100644 --- a/tests/test_sqlite_ledger.py +++ b/tests/test_sqlite_ledger.py @@ -5,7 +5,7 @@ from pathlib import Path from c3r.telemetry.sqlite_ledger import SqliteTraceLedger -from c3r.telemetry.trace_ledger import TraceLedger +from c3r.telemetry.trace_ledger import LedgerRecord, TraceLedger from tests.test_trace_ledger import trace @@ -14,8 +14,10 @@ def test_records_survive_restart_and_concurrent_appends(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "traces.sqlite3" ledger = SqliteTraceLedger(path) + def append(index: int) -> LedgerRecord: + return ledger.append(trace(f"run-{index}")) with ThreadPoolExecutor(max_workers=8) as pool: - list(pool.map(lambda i: ledger.append(trace(f"run-{i}")), range(100))) + list(pool.map(append, range(100))) ledger.close() reopened = SqliteTraceLedger(path) diff --git a/tests/test_staging_host.py b/tests/test_staging_host.py index 19936d0..f947472 100644 --- a/tests/test_staging_host.py +++ b/tests/test_staging_host.py @@ -6,8 +6,10 @@ class StagingHostTests(unittest.TestCase): - def test_staging_builder_cannot_enable_decisions_or_effects(self): - controller, factory = load_host_builder("c3r.staging_host:build")() + def test_staging_builder_cannot_enable_decisions_or_effects(self) -> None: + components = load_host_builder("c3r.staging_host:build")() + assert isinstance(components, tuple) + controller, factory = components self.assertFalse(controller.effect_execution_enabled) request = factory.build({"goal": "fixture", "current_subgoal": "check"}) self.assertEqual(request.estimates, {}) @@ -15,7 +17,7 @@ def test_staging_builder_cannot_enable_decisions_or_effects(self): self.assertEqual(outcome.reason, "C3R_DISABLED") self.assertIsNone(outcome.selected_action_id) - def test_staging_sink_has_no_row_store_or_cross_request_chain(self): + def test_staging_sink_has_no_row_store_or_cross_request_chain(self) -> None: trace = DecisionTrace( run_id="fixture_run", state_hash="a" * 64, access_level="internal", model_provider="fixture", candidate_ids=(), probabilities={}, diff --git a/tests/test_stateless_api.py b/tests/test_stateless_api.py index 1625f12..ef8bd58 100644 --- a/tests/test_stateless_api.py +++ b/tests/test_stateless_api.py @@ -3,14 +3,18 @@ import json import threading import unittest +from collections.abc import Callable, Mapping +from typing import cast from urllib.error import HTTPError from urllib.request import Request, urlopen from c3r.adapters.providers import ProviderAdapter, ProviderConfig, ProviderKind, TransportResponse +from c3r.deliberative.provider_bridge import ProviderDeliberator from c3r.host_factory import ReadOnlyRequestFactory from c3r.http_service import C3RHTTPServer from c3r.readiness import CachedReadiness from c3r.responses import ResponsesService +from c3r.runtime import RuntimeRequest from c3r.state_schema import ( ActionDefinition, ActionFamily, @@ -19,15 +23,29 @@ ValueEstimate, ) from c3r.system_one.advisory import AdvisoryFastPath -from c3r.system_one.clm_adapter import ClmAdapter +from c3r.system_one.clm_adapter import ClmAdapter, RankTransport from c3r.system_one.inference import SystemOneInference from tests.test_runtime import controller, request TOKEN = "stateless-test-token-with-at-least-thirty-two-characters" +def _ranked_fixture( + payload: Mapping[str, object], scores: tuple[float, float], +) -> Mapping[str, object]: + answers = payload["answers"] + if not isinstance(answers, list): + raise TypeError("fixture requires an answers list") + ranked: list[dict[str, object]] = [] + for option, score in zip(cast(list[object], answers), scores): + if not isinstance(option, str): + raise TypeError("fixture requires string answers") + ranked.append({"candidate": option, "prob": score}) + return {"model": "clm-latest", "ranked": ranked} + + class _Factory: - def build(self, payload): + def build(self, payload: Mapping[str, object]) -> RuntimeRequest: if payload.get("goal") != "Find record": raise ValueError("unknown goal") return request() @@ -49,7 +67,10 @@ def tearDown(self): self.server.server_close() self.thread.join(timeout=2) - def call(self, path, *, method="POST", token=TOKEN, payload=None): + def call( + self, path: str, *, method: str = "POST", token: str = TOKEN, + payload: Mapping[str, object] | None = None, + ): headers = {"Authorization": f"Bearer {token}"} data = None if payload is None else json.dumps(payload).encode() if data is not None: @@ -61,16 +82,22 @@ def call(self, path, *, method="POST", token=TOKEN, payload=None): except HTTPError as error: return error.code, json.load(error) - def configure_ranker(self, transport, *, enabled=True, system_one=True, ready=lambda: True): + def configure_ranker( + self, transport: RankTransport, *, enabled: bool = True, system_one: bool = True, + ready: Callable[[], bool] = lambda: True, + ) -> None: adapter = ClmAdapter("a" * 64, transport=transport) self.server.runtime, _ = controller(enabled=enabled, system_one=system_one, fast_path=AdvisoryFastPath(adapter)) self.server.system_one = SystemOneInference(adapter, readiness=ready) def test_typed_ranker_respects_disable_switches_without_provider_calls(self): - calls = [] + calls: list[Mapping[str, object]] = [] + def record(payload: Mapping[str, object]) -> Mapping[str, object]: + calls.append(payload) + return {} for enabled, system_one in ((False, True), (True, False)): - self.configure_ranker(lambda payload: calls.append(payload), enabled=enabled, + self.configure_ranker(record, enabled=enabled, system_one=system_one) for path in ("/v1/system-one", "/v1/c3r/rank"): status, _ = self.call(path, payload={"state": "test", "candidates": ["A", "B"]}) @@ -86,8 +113,9 @@ def test_model_availability_is_independent_of_system_two(self): self.assertTrue(body["data"][2]["available"]) def test_metadata_coalesces_health_checks_and_expires_cached_status(self): - now, calls = [0.0], [] - def probe(): + now: list[float] = [0.0] + calls: list[int] = [] + def probe() -> bool: calls.append(1) return len(calls) == 1 readiness = CachedReadiness(probe, clock=lambda: now[0]) @@ -109,7 +137,7 @@ def test_decide_and_rank_are_recommendation_only(self): self.assertEqual(body["authority_result"], "verified_not_committed") self.assertFalse(body["effect_executed"]) self.assertNotIn("confidence", body) - if path.endswith("rank") or path.endswith("system-one"): + if path.endswith(("rank", "system-one")): self.assertEqual(body["candidate_ranking"], []) self.assertTrue(body["abstained"]) @@ -127,11 +155,8 @@ def test_execute_and_untyped_responses_are_unavailable(self): self.assertEqual(body["error"], "not_implemented") def test_typed_system_one_answers_without_a_governed_action_catalog(self): - def rank(payload): - return {"model": "clm-latest", "ranked": [ - {"candidate": option, "prob": score} - for option, score in zip(payload["answers"], (0.8, 0.2)) - ]} + def rank(payload: Mapping[str, object]) -> Mapping[str, object]: + return _ranked_fixture(payload, (0.8, 0.2)) self.configure_ranker(rank) status, body = self.call("/v1/system-one", payload={ "model": "c3r-system-one", "state": "An invoice was charged twice", @@ -144,12 +169,51 @@ def rank(payload): self.assertFalse(body["calibrated"]) self.assertFalse(body["effect_executed"]) + def test_system_one_metadata_counts_actual_transport_attempts_per_question(self): + def rank(payload: Mapping[str, object]) -> Mapping[str, object]: + return _ranked_fixture(payload, (0.8, 0.2)) + self.configure_ranker(rank) + status, body = self.call("/v1/system-one", payload={ + "state": "test", "questions": {"first": {"type": "boolean"}, + "second": {"type": "boolean"}}}) + self.assertEqual(status, 200) + self.assertEqual((body["c3r"]["system_one_invocations"], + body["c3r"]["system_two_invocations"]), (2, 0)) + self.assertEqual(body["c3r"]["invocation_basis"], "adapter_transport_attempts") + + def test_failed_structured_decision_provider_is_counted_without_payload_details(self): + adapter = ProviderAdapter(ProviderConfig( + "local", ProviderKind.OPENAI_COMPATIBLE, "http://127.0.0.1:8000/v1", "model", None, + ), transport=lambda _url, _headers, _payload: TransportResponse( + 503, {"error": "PRIVATE_PROVIDER_FAILURE"}, 1)) + self.server.runtime, _ = controller(deliberative=True, deliberator=ProviderDeliberator(adapter)) + self.server.request_factory = ReadOnlyRequestFactory( + definitions=(ActionDefinition("reason", ActionFamily.DELIBERATE, "model", "plan", + RiskClass.READ_ONLY, ((),), ("local",), ("policy",), 1.0, 0.1),), + policy=AuthorityPolicy(frozenset({ActionFamily.DELIBERATE}), frozenset({RiskClass.READ_ONLY})), + estimate_source=lambda _: {"reason:0:local:policy": ValueEstimate(0.9, 0.1, 0.0, 0.1)}, + remaining_usd=1.0) + status, body = self.call("/v1/c3r/decide", payload={ + "goal": "Find record", "current_subgoal": "inspect"}) + self.assertEqual((status, body["reason"]), (200, "DELIBERATIVE_FAILURE")) + self.assertEqual((body["c3r"]["system_one_invocations"], + body["c3r"]["system_two_invocations"]), (0, 1)) + self.assertNotIn("PRIVATE_PROVIDER_FAILURE", json.dumps(body)) + + def test_failed_clm_transport_attempt_is_counted_without_private_diagnostics(self): + def failed(_payload: Mapping[str, object]) -> Mapping[str, object]: + raise OSError("PRIVATE_CLM_FAILURE") + self.configure_ranker(failed) + status, body = self.call("/v1/system-one", payload={ + "state": "test", "candidates": ["inspect", "abstain"]}) + self.assertEqual(status, 503) + self.assertEqual((body["c3r"]["system_one_invocations"], + body["c3r"]["system_two_invocations"]), (1, 0)) + self.assertNotIn("PRIVATE_CLM_FAILURE", json.dumps(body)) + def test_system_one_boolean_ranking_and_authority_rejection(self): - def rank(payload): - return {"model": "clm-latest", "ranked": [ - {"candidate": option, "prob": score} - for option, score in zip(payload["answers"], (0.25, 0.75)) - ]} + def rank(payload: Mapping[str, object]) -> Mapping[str, object]: + return _ranked_fixture(payload, (0.25, 0.75)) self.configure_ranker(rank) status, body = self.call("/v1/system-one", payload={ "state": "A duplicate charge", "questions": {"urgent": {"type": "boolean"}}, @@ -177,7 +241,7 @@ def test_system_one_rejects_total_option_overflow_and_provider_failure(self): def test_responses_returns_text_without_private_reasoning_or_external_effects(self): adapter = ProviderAdapter(ProviderConfig( "local", ProviderKind.OPENAI_COMPATIBLE, "http://127.0.0.1:8000/v1", "model", None, - ), transport=lambda *_: TransportResponse(200, { + ), transport=lambda _url, _headers, _payload: TransportResponse(200, { "choices": [{"message": {"content": "Check pending and settled charges.", "reasoning_content": "PRIVATE"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 12, "completion_tokens": 7}, @@ -199,12 +263,19 @@ def test_responses_returns_text_without_private_reasoning_or_external_effects(se self.assertNotIn("PRIVATE", json.dumps(body)) self.assertFalse(body["c3r"]["effect_executed"]) self.assertFalse(body["store"]) + self.assertEqual((body["c3r"]["system_one_invocations"], + body["c3r"]["system_two_invocations"]), (0, 1)) def test_positive_cvoc_generation_still_requires_independent_verification(self): - calls = [] + calls: list[int] = [] + def record( + _url: str, _headers: dict[str, str], _payload: dict[str, object], + ) -> TransportResponse: + calls.append(1) + return TransportResponse(200, {}, 0) adapter = ProviderAdapter(ProviderConfig( "local", ProviderKind.OPENAI_COMPATIBLE, "http://127.0.0.1:8000/v1", "model", None, - ), transport=lambda *_: calls.append(1)) + ), transport=record) runtime, _ = controller(deliberative=True, accepted=False) factory = ReadOnlyRequestFactory(definitions=(ActionDefinition( "DELIBERATE", ActionFamily.DELIBERATE, "compute", "generate", @@ -220,8 +291,8 @@ def test_positive_cvoc_generation_still_requires_independent_verification(self): self.assertEqual(calls, []) def test_hosted_text_generation_cannot_transfer_local_only_state(self): - calls = [] - def hosted_transport(*args): + calls: list[tuple[object, ...]] = [] + def hosted_transport(*args: object) -> TransportResponse: calls.append(args) return TransportResponse(200, {"choices": [{"message": {"content": "ready"}, "finish_reason": "stop"}]}, 10) @@ -242,8 +313,10 @@ def hosted_transport(*args): self.assertEqual(calls, []) def test_approved_hosted_responses_enforces_zero_retention_routing(self): - observed = [] - def hosted_transport(url, headers, payload): + observed: list[dict[str, object]] = [] + def hosted_transport( + url: str, headers: dict[str, str], payload: dict[str, object], + ) -> TransportResponse: observed.append(payload) return TransportResponse(200, {"choices": [{"message": {"content": "ready"}, "finish_reason": "stop"}], diff --git a/tests/test_trace_control_dry_run.py b/tests/test_trace_control_dry_run.py index 5a686d3..bc9cdfb 100644 --- a/tests/test_trace_control_dry_run.py +++ b/tests/test_trace_control_dry_run.py @@ -1,4 +1,5 @@ import unittest +from typing import cast from scripts.run_trace_control_dry_run import run @@ -8,8 +9,12 @@ def test_fixture_only_report_passes_without_claiming_deployed_controls(self): report = run() self.assertTrue(report["all_local_checks_passed"]) self.assertFalse(report["live_trace_collection_enabled"]) - self.assertEqual(len(report["checks"]), 9) - self.assertIn("deployed encryption and IAM", report["not_verified_by_this_run"]) + checks = report["checks"] + assert isinstance(checks, dict) + self.assertEqual(len(cast(dict[object, object], checks)), 9) + unverified = report["not_verified_by_this_run"] + assert isinstance(unverified, list) + self.assertIn("deployed encryption and IAM", cast(list[object], unverified)) if __name__ == "__main__": diff --git a/tests/test_trace_ledger.py b/tests/test_trace_ledger.py index 63046b2..7947121 100644 --- a/tests/test_trace_ledger.py +++ b/tests/test_trace_ledger.py @@ -3,7 +3,7 @@ from concurrent.futures import ThreadPoolExecutor from c3r.telemetry.trace import DecisionTrace -from c3r.telemetry.trace_ledger import TraceLedger +from c3r.telemetry.trace_ledger import LedgerRecord, TraceLedger def trace(run_id: str) -> DecisionTrace: @@ -52,8 +52,10 @@ def test_serialized_ledger_round_trips(self) -> None: def test_concurrent_appends_keep_one_valid_chain(self) -> None: ledger = TraceLedger() + def append(index: int) -> LedgerRecord: + return ledger.append(trace(f"run-{index}")) with ThreadPoolExecutor(max_workers=8) as pool: - list(pool.map(lambda index: ledger.append(trace(f"run-{index}")), range(100))) + list(pool.map(append, range(100))) self.assertEqual(len(ledger.records), 100) self.assertTrue(TraceLedger.verify(ledger.records))