From 934fc29c06dba1abf9fbd375d27c5a664d3fa344 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Fri, 23 Oct 2026 21:04:22 -0400 Subject: [PATCH 1/2] feat(cost): stage bounded session usage generations --- .../usage-attribution-state.contract.txt | 7 + docs/design/usage-attribution-state.md | 12 + docs/design/usage-generations-source.md | 100 +++++++++ docs/design/usage-generations.contract.txt | 54 +++++ docs/design/usage-generations.md | 10 + scripts/planes.py | 3 +- scripts/publish_work_cost.py | 50 ++++- scripts/retrieve_metric.py | 16 ++ scripts/usage_generation_budget.py | 96 ++++++++ scripts/usage_generation_codec.py | 144 ++++++++++++ scripts/usage_generation_publisher.py | 176 +++++++++++++++ scripts/usage_generation_reader.py | 141 ++++++++++++ scripts/usage_generation_step.py | 113 ++++++++++ scripts/usage_generation_terms.py | 56 +++++ scripts/usage_generation_wire.py | 126 +++++++++++ shapes/usage-generations.shapes.ttl | 47 ++++ tests/test_usage_generation_budget.py | 57 +++++ tests/test_usage_generation_codec.py | 93 ++++++++ tests/test_usage_generation_protocol.py | 206 ++++++++++++++++++ tests/test_usage_generation_step.py | 79 +++++++ tests/test_usage_generation_terms.py | 40 ++++ 21 files changed, 1620 insertions(+), 6 deletions(-) create mode 100644 docs/design/usage-attribution-state.contract.txt create mode 100644 docs/design/usage-attribution-state.md create mode 100644 docs/design/usage-generations-source.md create mode 100644 docs/design/usage-generations.contract.txt create mode 100644 docs/design/usage-generations.md create mode 100644 scripts/usage_generation_budget.py create mode 100644 scripts/usage_generation_codec.py create mode 100644 scripts/usage_generation_publisher.py create mode 100644 scripts/usage_generation_reader.py create mode 100644 scripts/usage_generation_step.py create mode 100644 scripts/usage_generation_terms.py create mode 100644 scripts/usage_generation_wire.py create mode 100644 shapes/usage-generations.shapes.ttl create mode 100644 tests/test_usage_generation_budget.py create mode 100644 tests/test_usage_generation_codec.py create mode 100644 tests/test_usage_generation_protocol.py create mode 100644 tests/test_usage_generation_step.py create mode 100644 tests/test_usage_generation_terms.py diff --git a/docs/design/usage-attribution-state.contract.txt b/docs/design/usage-attribution-state.contract.txt new file mode 100644 index 0000000..1163e70 --- /dev/null +++ b/docs/design/usage-attribution-state.contract.txt @@ -0,0 +1,7 @@ +Proposal42 REVISION — narrow SAME attribution encoding amendment, aegis-srm. Wu option1 accepted; Ian SAME scope. Original42 TTL2c06aab8/contract12240556 remain immutable history; this delta incorporates the original contract by reference and changes ONLY the conditional A request identity shape and row serialization requirement. No new class/engine feature, namespace migration or change to Malcolm budgets. New proposed property a:usageAttributionState (full IRI http://aegis.gastown.local/ontology/usageAttributionState), Camayoc owner, source shapes/usage-generations.shapes.ttl. On subjects with usageRequestId, exactly1 xsd:string in (attributed, no proven focus, overlapping focus), the exact parser outputs. Typed IRI, absent, duplicate and unknown enum values refuse. Attributed requires exactly1 attributedTo IRI; other two states require no attributedTo. Existing UsageRecord shape still validates referenced WorkItem type. C records without usageRequestId remain unchanged. No new property is emitted in production before normal proposal/source/profile acceptance and loaded-byte readback. + +The canonical row attribution slot is read from this explicit predicate, never inferred from absent links, label/comment, model-source or an offline ledger. Part/source digests preserve distinct no-proven-focus and overlapping-focus values. Counts/provider/model/model-source/observedAt/harness/request/session/actor/WorkItem links retain proposal42 definitions; each scalar must have one lossless RDF value of the expected kind/datatype. Reader refuses duplicate/wrong/missing/contradictory state before accepting a digest. Unattributed/overlap remain UNKNOWN for per-seed completeness; the enum does not fabricate ownership. A later attribution/reassignment changes row and full-generation digest but not stable request identity, and selected-generation/legacy exclusion removes prior allocations. Empty generation and all prior limitations/rollback/cadence/storage/runtime gates remain exactly42. + +Validation: private actual loaded core+breakdown+revised candidate, old C positive unchanged; actual publisher-derived records augmented only with proposed request identity+explicit state in the PRIVATE fixture, native /knot and single lossless /query round-trip, reconstruct the canonical row/digest. Both previously indistinguishable source arms must now have distinct RDF and distinct matching reader digests. Absence/duplicate/wrong type/wrong enum and contradictions refuse. These are candidate controls, not production emission or partition source approval. Ian's review is delta-only; pending proposal42 is not retroactively accepted/loaded by this amendment. + +Original proposal42 exact TTL SHA2562c06aab8d6b6931ef720047ce521eb38b4e68416daa521989e26b44dd8790895, contract122405564b14e98a6d94262234301915f24dbea5e75a14ed7bb5a4e25c42d23b. Native emitter/reader pairs and controls in same archive; last full candidate31controls includes refused subject absence; prior25positive native controls preserved. New source implementation remains held for Ian delta approval. \ No newline at end of file diff --git a/docs/design/usage-attribution-state.md b/docs/design/usage-attribution-state.md new file mode 100644 index 0000000..f8fe7fd --- /dev/null +++ b/docs/design/usage-attribution-state.md @@ -0,0 +1,12 @@ +# Attribution state amendment + +The [proposal 43 submitted rationale](usage-attribution-state.contract.txt) +is retained byte for byte, including its proposal 42 lineage paragraph. +The exact revised TTL is `shapes/usage-generations.shapes.ttl`. + +The conditional A request shape requires one explicit attribution-state string. +The three parser states round-trip independently; absent, duplicate, wrong-kind, +unknown-enum and contradictory states refuse. Old C records remain compatible. + +Proposal acceptance, normal profile deployment and runtime enrollment remain +separate from this reviewed source change. diff --git a/docs/design/usage-generations-source.md b/docs/design/usage-generations-source.md new file mode 100644 index 0000000..0c6690c --- /dev/null +++ b/docs/design/usage-generations-source.md @@ -0,0 +1,100 @@ +# Staged usage generation source + +This implements the source-only SRM option A decision. It does not enroll a +producer, accept/load a proposal, create a production graph, or declare runtime +acceptance. Proposal 42 and its approved proposal 43 attribution delta remain +separate from normal source review and deployment. The exact revised profile +is `shapes/usage-generations.shapes.ttl`; the original contract and the narrow +amendment are retained in the adjacent design documents. + +## Entry points and scope + +`publish_work_cost.py --post --generation-policy POLICY.json` explicitly selects +A for that policy's session when C refuses its item count, or when A already +owns its generation. Omitting the option retains C. C and A share the same +producer lock, pending-write fence and hourly/minute write ledger when enrolled. +A may not publish without the existing observed-record read grant, an exact +loaded profile match, an explicit frozen-legacy declaration and a runtime +acceptance reference. No policy is installed by this source change. + +The policy schema has exactly `enabled`, `producer`, `harness`, `session`, +`legacy_frozen`, and `runtime_acceptance`. An enabled policy requires both +booleans true, an existing reviewed producer IRI, an exact native session ID, +a known harness, and the separately reviewed runtime acceptance reference. +These fields are trusted deployment configuration, not a substitute for the +runtime evidence or permission to manufacture it. The normal policy author +must bind that evidence; a nonempty string alone is not operational authority. + +One generation contains disjoint stable-request parts plus its manifest. +A part is at most 40 requests and 64 KiB of final encoded HTTP body. A metadata +fragment contains zero usage records, is bounded by the same 64 KiB write +budget, and has its own tick. It records the complete expected nonempty-part +set; it is not counted as a request partition. This permits empty correction +generations without inventing a new class or claiming an empty token source. +The parser must still establish the supplied input has no errors. + +Graph creation and kind-labeling each occupy their own tick. Kind labels are +`knowledge`; this source assigns no trust rank, authority or signing identity. +Part staging uses at most three requests: direct canonical WorkItem preflight, +write, exact lossless readback. Graph-local reference stubs contain only a +preflight-proven WorkItem type and the identifier as label, never a guessed +canonical title. They live only in the generation graph and are excluded from +cost rows. No canonical tracker graph or original identity fact is rewritten. + +The manifest is written and verified only after every request part has a durable +receipt. Activation is a marker write and one graph-bound complete-set query, +at most two requests. The reader validates graph identity, all required term +kinds/cardinalities, canonical JSON, horizon, source/manifest/part digests, +counts, request identity, correction lineage and attribution state. Unexpected +facts, missing rows, mixed generations, legacy graphs and truncation refuse as +UNKNOWN. Source-log completeness and global per-seed coverage remain separate. + +## Load and failure bounds + +The shared lane admits at most one C-or-A write per minute and 60 writes/hour. +A body bytes are at most 3.75 MiB/hour; combined C+A bytes cannot exceed the +measured prior worst-case hourly envelope. Aggregate planned/staged unselected +body receipts are capped at 64 MiB. Unknown state is never reset to zero. +A single over-budget record or metadata envelope refuses before graph writes. +An unchanged selected source is not restaged. + +The planner retains one selected generation and one staging/inactive slot plus +the frozen legacy source. After a correction selects the next generation, the +old A graph stays in the inactive slot. Another generation refuses until the +separately governed retention step frees it. No retirement, deletion or hidden +storage growth is implemented here. This conservative stop is an explicit +operational dependency, not permission to delete retained graphs. + +Pending is persisted before every write; an uncertain response or mismatched +readback keeps it. There is no same-tick retry or automatic pending clear. A +verified receipt can recover only a lost local cursor update by exact body +hash, without resending. Existing cooldown rules and the shared rate fence +remain in force; slow calls retain the longer backoff. + +## Read consumers + +The opt-in `session_usage/work_cost_generation` retrieval method accepts exactly +producer, harness, session, horizon and consumer (`st-cost` or `brief`). It uses +one cached attempt fence per consumer across sessions, with at most one query +per five minutes. No new cadence or background poller is introduced. The query +returns selector and generation facts in one response, with explicit STR, +isIRI, DATATYPE and LANG columns and the normal truncation indicator. It does +not confuse an IRI-looking literal with an IRI or silently accept a capped +standard-format result. + +A failed or rate-limited build returns UNKNOWN with any prior complete snapshot +explicitly separated as `last_complete`. It never substitutes it as fresh cost. +The pre-existing metrics, Brief cohort admission and C coverage limitations are +not bypassed. This source adds no production method declaration or variable flip. + +## Rollback and remaining acceptance + +Pause generation publication and retain the generation-aware reader, pending +bodies, manifests, receipts and graphs. A pointer rollback may expose only +explicitly stale/UNKNOWN older data. Never resume C concurrently for the same +session, clear pending, or delete stage/legacy graphs as a rollback shortcut. + +Runtime still requires the governed profile disposition/load/readback, a +real-shaped replay at production-copy size with the signed latency limits, +the fleet 408 gate or explicit lead ruling, then the matched after-watch. +The small private native controls are source correctness evidence only. diff --git a/docs/design/usage-generations.contract.txt b/docs/design/usage-generations.contract.txt new file mode 100644 index 0000000..4cf6f93 --- /dev/null +++ b/docs/design/usage-generations.contract.txt @@ -0,0 +1,54 @@ +SRM production shape CANDIDATE v1 — gate(c), review requested, NOT authorization + +Task aegis-srm. Owner: Camayoc producer/profile, proposed source path shapes/usage-generations.shapes.ttl beside existing shapes/usage-breakdown.shapes.ttl. No Quipu engine shape or partition implementation. This archive is the exact review candidate; accepted bytes must later land in Camayoc under normal source review and reach the loaded server before any new typed producer write. /propose is pending review only, never deployment. Existing C remains unchanged. + +Namespace and reuse + +The immutable http://aegis.gastown.local/ontology/ namespace is taken from Camayoc's CURRENT loaded camayoc-core (authenticated get, loaded2026-10-23T22:07:31Z) and existing usage-breakdown source. m5s repository/website moves do not migrate this IRI. Do not substitute intigraph.ai IDs or synthetic srm.test predicates. Fresh vocabulary lists171 classes incl UsageRecord; it does NOT establish property semantics. The 2 NEW classes and 12 NEW properties below are EXPLICIT PROPOSALS, not claimed already governed or allowed for production writes. Ian owns acceptance. Example producer/instance IRIs in fixtures are test values, not a registered producer identity or new live authority. Final producer identity must be an existing reviewed software/producer identity, pinned with its source artifact before enrollment. + +Terms (a: abbreviates the immutable namespace above; all owned by this Camayoc profile) + +| Term | Reuse/new | Target and kind/cardinality | Semantics | +| --- | --- | --- | --- | +| a:UsageRecord | REUSE loaded class | actual request usage only | Never type a manifest/selector as UsageRecord. Existing token/model/provider/observedAt/attributedTo facts remain unchanged. | +| a:Session | REUSE loaded class | existing session object | Preserve existing C identity/shape; no migration. | +| a:inSession | REUSE loaded core property | generation/selector/request identity: IRI exactly1 | Exact session identity, not seed assignment. | +| a:usageHarness | REUSE usage-breakdown property | exactly1 string claude/codex on new profile subjects | Harness is part of identity; identical provider request IDs across harnesses cannot coalesce. Existing old UsageRecords need not gain new metadata. | +| a:UsageGeneration | NEW class | manifest node | Immutable proposed generation metadata, NOT a permanent complete assertion. | +| a:UsageGenerationSelector | NEW class | per-producer/harness/session selector | Atomic small snapshot points at one selected generation and digest. | +| a:usageProducer | NEW | generation+selector: IRI exactly1 | Stable reviewed producer identity, not whichever agent is currently assigned the seed. | +| a:usageProfileVersion | NEW | generation+selector: string exactly1, srm-generation-v1 | Encoding/semantic version; unknown version => UNKNOWN. | +| a:usageHorizon | NEW | generation+selector: UTC second string exactly1 | Source observation upper bound at captured input digest; not a declaration of global log completeness. Strict calendar validity checked by consumer; SHACL lexical pattern alone is insufficient. | +| a:usageGraph | NEW | generation: IRI exactly1 | Exact registered selected data graph; no ROOT fallback or widening. | +| a:usageSourceDigest | NEW | generation: lowercase hex SHA256 string exactly1 | Hash of canonical complete input request records through horizon, including absence/unknown and attribution corrections. | +| a:usageManifestDigest | NEW | generation+selector: lowercase hex SHA256 string exactly1 | Digest of exact canonical manifest bytes defined below. Selector and manifest must agree in the SAME response. | +| a:usageManifestJSON | NEW | generation: string exactly1 | Canonical complete expected partition descriptor set and record count. Not a claim those partitions actually exist. | +| a:usagePartitionCount | NEW | generation: integer exactly1, >=0 and <=recordCount | Number of expected nonempty disjoint partitions; zero only for an empty-generation correction. | +| a:usageRecordCount | NEW | generation: integer exactly1, >=0 | Distinct expected request IDs; compare descriptor sum and actual facts, never count duplicate labels as records. | +| a:usagePreviousGeneration | NEW | generation: IRI0..1 | Predecessor for correction lineage; never a union/read fallback. Initial absent, then exact selected predecessor. | +| a:usageSelectedGeneration | NEW | selector: IRI exactly1 | Manifest/generation identity; no arbitrary extra generation selections. | +| a:usageRequestId | NEW | actual UsageRecord carrying this property: nonempty string exactly1 | Exact native request ID. Conditional targetSubjectsOf means old records without it keep their original shape behavior. New A records require it by producer/reader contract. | + +SHACL details are exact in usage-generations.proposed.ttl: URI kinds, count datatypes/bounds, mandatory metadata, lowercase64hex digests, harness/version enumeration, cardinality1 and optional predecessor0..1. Types Session/UsageRecord and existing breakdown properties are NOT redefined. No new part class is needed: the expected part set is a canonical JSON value. This avoids pretending 40 metadata entries in one selector can fit the2KiB marker budget. Existing token shape and core constraints remain loaded together with this candidate in the private paired controls. + +Identity, canonical encoding, horizon and corrections + +Request key is the EXACT tuple [harness, native-session-id, native-request-id], independent of agent/model/seed attribution. A request ID is compared byte-exact after UTF8 decoding; no casefold, trim or Unicode normalization; ill-formed Unicode refuses. Candidate A instance names are SHA256 of the canonical tuple under the existing cost namespace, not a rewrite of existing C Session/UsageRecord IRIs. Proposed A request path cost/request/; generation path cost/generation/; selector path cost/selector/. These NEW instance paths are candidate conventions, not minted production entities. inSession remains the existing Session IRI; raw session ID is in the digest-bound manifest. A reader MUST compare it against the exact selected session identity route, not derive it from an ambiguous label. Multi-harness collisions remain separated by the tuple. + +Canonical bytes: UTF8 JSON, object keys ASCII sorted, separators comma/colon without whitespace, strings use JSON escapes for control/quote/backslash, non-ASCII scalar values emitted literally; reject duplicate keys, nonfinite numbers, floats, surrogate codepoints, unknown fields or alternate byte serialization. JSON schema keys are exactly version, producer, harness, session, horizon, source_sha256, previous, parts, records. Counts are nonnegative safe JSON integers. version is srm-generation-v1; producer is exact IRI string; harness is claude/codex; session exact string; horizon strict real UTC YYYY-MM-DDTHH:MM:SSZ; source_sha256 lowercase64hex; previous is null or exact predecessor IRI. parts is ordinal-sorted contiguous0..N-1, unique descriptors [ordinal, record-set-sha256, distinct-record-count, encoded-knot-body-bytes]. records equals sum of descriptor record counts. Empty set has parts=[]/records=0. Each nonempty part is <=40 records AND encoded HTTP JSON knot body<=65536 bytes; marker body<=2048 bytes. Digest=lowercase SHA256 of these EXACT bytes. The fixture canonical JSON/digest are archived. + +Part record-set digest uses the same canonical JSON encoding over request-key-sorted arrays. Each array binds exact request identity, all parser token fields as canonical decimal strings or explicit null, provider/model/model-source, observed timestamp, attribution state, sorted exact WorkItem IRIs, and actor. Missing/unknown is NOT zero. No float conversion. Full source digest covers the complete canonical record array, including late/removed/reassigned records; the manifest includes its hash. Any correction generates a new complete expected set and generation digest, never a seed-keyed partition migration. Part body-byte length is measured from final JSON HTTP bytes, not Turtle characters. The part digest is canonical parsed record data, NOT raw Turtle serialization, so graph-IRI/digest embedding does not create a hash cycle. This is a profile contract to implement later, not producer source in this task. + +The existing C session IRI omits harness; it is NOT silently declared collision-safe. A readers must bind full tuple and reject ambiguous legacy mapping; legacy C is excluded after scoped verified selection. Empty generation is allowed to remove all prior allocations; no native graph deletion needed. If an A session shrinks back within C eligibility, concurrent C/A publication is forbidden; that transition needs the same reviewed selector/legacy discipline, not a second publisher. + +SHACL vs operational validation + +Shapes cannot prove sums inside JSON, matching digest, receipt existence, graph immutability, real horizon completeness, freshness or cross-node producer/session/horizon equality. The private fixture oracle demonstrates these as REQUIRED reader/producer checks; it is NOT partition source and NOT a general validator implementation. Inconsistent partitionCount>recordCount is refused by actual SHACL; a different but plausible total can remain SHACL-valid and MUST fail canonical manifest/count validation. No false claim that SHACL enforces arithmetic over arbitrary JSON. One reader response must include selector identity/digest, manifest, all expected part receipts and facts in graph scope. Exact typed terms and lexical values must be retained for digest reconstruction; the legacy lossy query JSON shape must NOT be treated as proof of RDF kinds (use an admitted lossless representation or refuse UNKNOWN). This is a remaining implementation/read contract, not permission to weaken exactness. + +No complete=true field. Missing/mixed/malformed/tampered/duplicate/truncated/unknown metadata => UNKNOWN. Source remains immutable after sealed verification; pointer rollback exposes explicitly stale prior data only. Old whole-session publisher must be frozen before selection, never run concurrently. Do not delete legacy/stage data during rollback. Pending/uncertain write -> exact readback before any resend, no automatic replay/reset. + +Compatibility, budget and rollback + +Baseline: production loaded core hash0b689c1c1a19f0f9ebf4ffff0400694c666bee5499dc9b030d83679f3b5f9970; usage-breakdown source hash47953d51fefc8eb33fdca2999031131d857049525855630b7ef57f9b852e869b. Candidate changes no source/runtime publisher. First normal source landing after Ian approval; normal shape load/readback must expose both new classes and exact named profile before new typed ingress. Accepted proposal alone is not durable deployment. Removal rollback only after A publication is paused and no new type is emitted; retain profiles while readers/pending writes need them. C has no new mandatory fields and remains compatible. + +Malcolm gate(b) c23 signs A SOURCE only: serial-trickle cadence-shared ONE C-or-A write/tick, staging<=3 requests, activation<=2, create/label own tick; part<=64KiB/40records; marker<=2KiB; <=60writes/hour; one in-progress+one selected+legacy/session; staged-body-receipts<=64MiB; no unchanged restage; one graph-bound consumer read/build <=1/5min. No cap raise, extra cadence, runtime admission, purge or source workaround. Runtime still requires real-shaped prod-size replay,408gate and matched after-watch. Synthetic proof120B records is ~20x smaller than real~2.4KiB records; its timings do not transfer. diff --git a/docs/design/usage-generations.md b/docs/design/usage-generations.md new file mode 100644 index 0000000..bd41553 --- /dev/null +++ b/docs/design/usage-generations.md @@ -0,0 +1,10 @@ +# Usage generation contract + +The [proposal 42 contract](usage-generations.contract.txt) is retained byte for +byte as the normative starting contract. The approved +[proposal 43 attribution amendment](usage-attribution-state.md) narrows its +conditional request shape; it does not load or activate a profile. + +See [source behavior and remaining gates](usage-generations-source.md) for +the implementation boundaries, source-only evidence and rollback. Neither +a valid manifest nor source installation proves complete per-seed accounting. diff --git a/scripts/planes.py b/scripts/planes.py index b090dbd..5a9ce20 100755 --- a/scripts/planes.py +++ b/scripts/planes.py @@ -147,7 +147,8 @@ def provenance_headers(client: str, env=None) -> dict[str, str]: def _post(path: str, body: dict, *, client: str, form: bool = False, response_limit: int | None = None) -> dict: - if response_limit is not None and (type(response_limit) is not int or not 1 <= response_limit <= 262144): + maximum_response = 8 * 1024 * 1024 if client == 'camayoc-cost-generation' else 262144 + if response_limit is not None and (type(response_limit) is not int or not 1 <= response_limit <= maximum_response): raise PlaneError("Invalid explicit response budget") req = urllib.request.Request( f"{SERVER}{path}", diff --git a/scripts/publish_work_cost.py b/scripts/publish_work_cost.py index c7e614c..5c6a285 100644 --- a/scripts/publish_work_cost.py +++ b/scripts/publish_work_cost.py @@ -171,8 +171,13 @@ def record_budget(history, receipt): def publish(result, actor, state_path, *, push_status=False, review_only=False, - scheduler_only=None): + scheduler_only=None, generation_policy=None, _lane_locked=False): """Every attempt leaves a receipt and status metrics, including failures.""" + if not _lane_locked: + from usage_generation_step import locked + with locked(state_path): + return publish(result,actor,state_path,push_status=push_status,review_only=review_only, + scheduler_only=scheduler_only,generation_policy=generation_policy,_lane_locked=True) receipt_path = state_path.with_suffix('.receipt.json') receipt = {'attempted_at': time.time(), 'status': 'UNKNOWN', 'requests': 0, 'request_seconds': 0, 'items_per_snapshot': [], 'records_per_snapshot': [], @@ -215,8 +220,8 @@ def publish(result, actor, state_path, *, push_status=False, review_only=False, receipt[field] = previous.get(field, []) receipt['dimensions_source'] = 'previous_attempt' raise OperatorAction('BACKOFF: previous request budget cooldown has not elapsed') - answer = _publish(result, actor, state_path, receipt) - receipt['status'] = 'OK' + answer = _publish(result, actor, state_path, receipt, generation_policy=generation_policy, _lane_locked=_lane_locked) + receipt['status'] = 'UNKNOWN' if receipt.get('unknown_reason')=='staging' else 'OK' return answer except OperatorAction as exc: receipt.update(status='STALLED' if state_path.with_suffix('.pending.json').exists() else 'OPERATOR_ACTION', @@ -289,8 +294,35 @@ def publish(result, actor, state_path, *, push_status=False, review_only=False, raise RuntimeError('cost status metrics were not delivered: ' + why) -def _publish(result, actor, state_path, receipt): +def _publish(result, actor, state_path, receipt, *, generation_policy=None, shared_budget=False, _lane_locked=False): require_read_grant() + if generation_policy is not None: + from usage_generation_publisher import policy_check, tick + policy_check(generation_policy) + rows=result.get('records',[]) + matches=all(r.get('harness')==generation_policy['harness'] and r.get('session')==generation_policy['session'] for r in rows) + items={r.get('bead') for r in rows if r.get('attribution')=='attributed'} + journal=state_path.with_suffix('.generations.json') + owned=False + if journal.exists(): + from usage_generation_publisher import load_state + from usage_generation_codec import digest + state=load_state(journal) + owned=digest([generation_policy['producer'],generation_policy['harness'],generation_policy['session']]) in state['sessions'] + if matches and (len(items)>MAX_WORK_ITEMS or owned): + return tick(result,actor,state_path,receipt,generation_policy,lane_locked=_lane_locked) + return _publish(result,actor,state_path,receipt,shared_budget=True,_lane_locked=_lane_locked) + generation_journal=state_path.with_suffix('.generations.json') + if generation_journal.exists(): + from usage_generation_publisher import load_state + journal=load_state(generation_journal) + identities={(r.get('harness'),r.get('session')) for r in result.get('records',[])} + for entry in journal['sessions'].values(): + owner=entry.get('progress',{}).get('plan',{}).get('manifest',{}) if entry.get('progress') else entry.get('selected',{}) + if not isinstance(owner,dict) or not owner.get('harness') or not owner.get('session'): + raise ValueError('UNKNOWN: unreadable generation ownership; no legacy fallback') + if (owner['harness'],owner['session']) in identities: + raise ValueError('legacy whole-session publication frozen for generation-owned session') if state_path.with_suffix('.pending.json').exists(): raise ValueError('previous graph write indeterminate; reconcile pending snapshot before retry') state = json.loads(state_path.read_text()) if state_path.exists() else {} @@ -382,7 +414,14 @@ def post(endpoint, body): marker = state_path.with_suffix('.pending.json') if marker.exists(): raise ValueError('previous graph write indeterminate; reconcile pending snapshot before retry') + if shared_budget: + from usage_generation_budget import admit + ledger_path=state_path.with_suffix('.writes.json') + ledger=json.loads(ledger_path.read_text()) if ledger_path.exists() else [] + next_ledger=admit('whole',len(json.dumps(body,sort_keys=True).encode()),len(records),3,time.time(),ledger) marker.write_text(json.dumps(body, sort_keys=True)) + if shared_budget: + _atomic(ledger_path,json.dumps(next_ledger,sort_keys=True)) response = post('/knot', body) if response.get('conforms') is False or not response.get('tx_id'): raise ValueError('graph refused cost snapshot') @@ -410,6 +449,7 @@ def post(endpoint, body): def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument('method', type=Path, nargs='?') + ap.add_argument('--generation-policy',type=Path,help='Explicit reviewed A runtime enrollment; absent keeps C unchanged') ap.add_argument('--actor', required=True) ap.add_argument('--state', type=Path, required=True) mode = ap.add_mutually_exclusive_group() @@ -435,7 +475,7 @@ def main(): if method.get('system') != 'session_usage' or method.get('query') != 'work_cost': raise ValueError('expected session_usage/work_cost method') result = retrieve(method['params']) - answer = publish(result, args.actor, args.state, push_status=args.publish_status) if args.post else [b for b, _ in snapshots(result, args.actor)] + answer = publish(result, args.actor, args.state, push_status=args.publish_status, generation_policy=json.loads(args.generation_policy.read_text()) if args.generation_policy else None) if args.post else [b for b, _ in snapshots(result, args.actor)] print(json.dumps(answer, sort_keys=True)) return 0 except (OSError, ValueError, RuntimeError) as exc: diff --git a/scripts/retrieve_metric.py b/scripts/retrieve_metric.py index ca1e4b4..33711d7 100755 --- a/scripts/retrieve_metric.py +++ b/scripts/retrieve_metric.py @@ -26,6 +26,22 @@ def execute(method: dict[str, Any], timeout: float = 5.0) -> dict[str, Any]: return {"status": "retrieved", "system": system, "result": retrieve(params)} except (ValueError, TypeError, KeyError) as exc: return {"status": "invalid", "system": system, "error": str(exc)} + if system == "session_usage" and query == "work_cost_generation": + from pathlib import Path + import time + from usage_generation_reader import consume + if set(params) != {"producer", "harness", "session", "horizon", "consumer"} or params.get("consumer") not in ("st-cost", "brief"): + return {"status":"invalid", "system":system, "error":"invalid generation reader scope"} + # One persistent attempt fence per consumer, across every session and + # generation. A caller cannot multiply rate by choosing new filenames. + directory=Path(os.environ.get("CAMAYOC_COST_READ_STATE",Path.home()/".local/state/camayoc/cost-reads")) + try: + directory.mkdir(parents=True,exist_ok=True) + value=consume(producer=params["producer"],harness=params["harness"],session=params["session"], + at=params["horizon"],state_path=directory/(params["consumer"]+".json"),now=time.time()) + return {"status":"retrieved" if value["status"]=="COMPLETE" else "unknown", "system":system, "result":value} + except (ValueError,TypeError,KeyError,OSError): + return {"status":"unknown", "system":system, "error":"generation validation unavailable"} if system != "prometheus": return {"status": "unsupported", "system": system} endpoint = params.get("endpoint") diff --git a/scripts/usage_generation_budget.py b/scripts/usage_generation_budget.py new file mode 100644 index 0000000..9fb636e --- /dev/null +++ b/scripts/usage_generation_budget.py @@ -0,0 +1,96 @@ +"""Pure source-contract guards for serial-trickle cost publication (SRM gate b). + +The caller must hold the existing producer lock, persist a pending write before +networking, and share this ledger between C and A. These checks perform no I/O. +""" +import math + +from usage_generation_codec import (PART_BYTES, PART_RECORDS, MARKER_BYTES, + STAGED_BYTES, WRITES_HOUR, count) + +HOUR_SECONDS = 3600 +MIN_TICK_SECONDS = 60 +A_BYTES_HOUR = 60 * PART_BYTES +TOTAL_BYTES_HOUR = 60 * 1_640_000 +MAX_REQUESTS = {'part': 3, 'marker': 2, 'graph': 2, 'graph_label': 2, 'whole': 3} + + +def timestamp(value): + if type(value) not in (int,float) or not math.isfinite(value) or value < 0: + raise ValueError('invalid budget clock') + return value + + +def admit(kind, body_bytes, records, requests, now, ledger, staged_bytes=0): + """Return the exact next ledger; never reset/drop unknown history on error.""" + if kind not in MAX_REQUESTS: + raise ValueError('unknown cost write class') + count(body_bytes); count(records); count(requests); count(staged_bytes) + timestamp(now) + if requests > MAX_REQUESTS[kind] or requests < 1: + raise ValueError('per-tick request budget') + if kind == 'part' and (body_bytes > PART_BYTES or records > PART_RECORDS): + raise ValueError('part budget') + if kind == 'marker' and body_bytes > MARKER_BYTES: + raise ValueError('marker budget') + if staged_bytes > STAGED_BYTES: + raise ValueError('aggregate staged-byte budget') + if not isinstance(ledger,list): + raise ValueError('unreadable write ledger') + previous = -1 + for row in ledger: + if not isinstance(row,dict) or set(row) != {'at','bytes','kind'} or row['kind'] not in MAX_REQUESTS: + raise ValueError('invalid write receipt') + stamp = timestamp(row['at']); count(row['bytes']) + if stamp < previous or stamp > now: + raise ValueError('clock regression/unsorted receipts') + previous = stamp + if ledger and now - ledger[-1]['at'] < MIN_TICK_SECONDS: + raise ValueError('one C-or-A write per scheduled minute') + window = [dict(row) for row in ledger if row['at'] > now-HOUR_SECONDS] + if len(window) >= WRITES_HOUR: + raise ValueError('cost writes/hour budget') + if sum(row['bytes'] for row in window)+body_bytes > TOTAL_BYTES_HOUR: + raise ValueError('C+A body/hour budget') + if kind != 'whole' and sum(row['bytes'] for row in window if row['kind']!='whole')+body_bytes > A_BYTES_HOUR: + raise ValueError('A body/hour budget') + return window + [{'at':now,'bytes':body_bytes,'kind':kind}] + + +def admit_session(*, c_items, source_digest, previous, staged_bytes, body_bytes): + """No second in-progress generation, unchanged restage or unbounded staging.""" + count(c_items); count(staged_bytes); count(body_bytes) + if c_items <= 8: + raise ValueError('A only serves C-refused oversized sessions') + if not isinstance(source_digest,str) or len(source_digest)!=64 or any(c not in '0123456789abcdef' for c in source_digest): + raise ValueError('invalid session source digest') + if not isinstance(previous,dict) or set(previous)-{'selected','in_progress','legacy'}: + raise ValueError('unknown generation registry') + for role in ('selected','in_progress','legacy'): + if previous.get(role) is not None and not isinstance(previous[role],dict): + raise ValueError('unknown retained generation') + selected = previous.get('selected') + if selected and selected.get('source_digest') == source_digest: + raise ValueError('unchanged session is not restaged') + progress = previous.get('in_progress') + if progress and progress.get('status') not in ('sealed','abandoned'): + raise ValueError('previous generation is neither sealed nor explicitly abandoned') + if staged_bytes+body_bytes > STAGED_BYTES: + raise ValueError('aggregate staged-byte budget') + # No retirement/delete operation is authorized by this guard. + return True + + +def admit_read(*, now, previous, graph, requests=1, response_bytes=0): + """Consumer-owned persisted attempt fence, one bounded graph read/build/5min.""" + from usage_generation_codec import iri, READ_BYTES + timestamp(now); iri(graph); count(requests); count(response_bytes) + if requests != 1 or response_bytes > READ_BYTES: + raise ValueError('single bounded generation query required') + if previous is not None: + if not isinstance(previous,dict) or set(previous)!={'at','graph'}: + raise ValueError('unknown consumer receipt') + before=timestamp(previous['at']); iri(previous['graph']) + if now < before+300: + raise ValueError('consumer cadence/clock regression') + return {'at':now,'graph':graph} diff --git a/scripts/usage_generation_codec.py b/scripts/usage_generation_codec.py new file mode 100644 index 0000000..1de6d0b --- /dev/null +++ b/scripts/usage_generation_codec.py @@ -0,0 +1,144 @@ +"""Proposal42 canonical accounting envelopes; pure, with no graph/network writes.""" +from __future__ import annotations + +from datetime import datetime +import hashlib +import json +import re +from urllib.parse import urlsplit + +from ingest_session_usage import BASE, ONTOLOGY + +VERSION = 'srm-generation-v1' +PART_BYTES = 65536 +PART_RECORDS = 40 +MARKER_BYTES = 2048 +STAGED_BYTES = 64 * 1024 * 1024 +WRITES_HOUR = 60 +READ_BYTES = 8 * 1024 * 1024 +HEX = re.compile(r'^[0-9a-f]{64}$') +FIELDS = ('version', 'producer', 'harness', 'session', 'horizon', 'source_sha256', + 'previous', 'parts', 'records') +COUNTS = ('input_uncached', 'cache_read_input', 'cache_write_input', 'output') + + +def canonical(value): + def valid(v): + if v is None or type(v) is bool: + return + if type(v) is int: + if abs(v) > 2**53 - 1: + raise ValueError('unsafe JSON integer') + elif isinstance(v, str): + v.encode('utf-8', 'strict') + elif isinstance(v, list): + for x in v: + valid(x) + elif isinstance(v, dict) and all(isinstance(k, str) and k.isascii() for k in v): + for x in v.values(): + valid(x) + else: + raise ValueError('noncanonical JSON value') + valid(value) + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')) + + +def digest(value): + return hashlib.sha256(canonical(value).encode()).hexdigest() + + +def iri(value): + if not isinstance(value, str) or any(c in value for c in '<>"{}|^`\\\n\r\t '): + raise ValueError('invalid IRI') + parsed = urlsplit(value) + if parsed.scheme not in ('http', 'https') or not parsed.netloc: + raise ValueError('IRI must be absolute HTTP(S)') + value.encode('utf-8', 'strict') + return value + + +def horizon(value): + if not isinstance(value, str) or not re.fullmatch(r'\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ', value): + raise ValueError('invalid horizon') + datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ') + return value + + +def text(value): + if not isinstance(value, str) or not value: + raise ValueError('missing identity/string') + value.encode('utf-8', 'strict') + return value + + +def count(value): + if type(value) is not int or value < 0: + raise ValueError('invalid count') + return value + + +def record_row(record): + """Stable tuple, no seed-assignment identity and no float token conversions.""" + harness = record['harness'] + if harness not in ('claude', 'codex'): + raise ValueError('unknown harness') + key = [harness, text(record['session']), text(record['id'])] + counts = [None if record['counts'].get(k) is None else str(count(record['counts'][k])) for k in COUNTS] + attribution = record['attribution'] + if attribution not in ('attributed', 'no proven focus', 'overlapping focus'): + raise ValueError('unknown attribution') + items = [iri(ONTOLOGY + text(record['bead']))] if attribution == 'attributed' else [] + # Provider derives from the parser harness, never an environment default. + provider = 'anthropic' if harness == 'claude' else 'openai' + model = None if record.get('model') is None else text(record['model']) + model_source = None if record.get('model_source') is None else text(record['model_source']) + return [key, str(count(record['tokens'])), counts, provider, model, + model_source, text(record['at']), attribution, items, text(record['agent'])] + + +def record_iri(row): + return BASE + 'request/' + digest(row[0]) + + +def validate_manifest(raw, *, producer, harness, session, at): + def pairs(values): + out = {} + for k,v in values: + if k in out: + raise ValueError('duplicate JSON key') + out[k] = v + return out + m = json.loads(raw, object_pairs_hook=pairs) + if not isinstance(m, dict) or set(m) != set(FIELDS) or canonical(m) != raw: + raise ValueError('manifest is not exact canonical schema') + if (m['version'],m['producer'],m['harness'],m['session'],m['horizon']) != (VERSION,iri(producer),harness,text(session),horizon(at)): + raise ValueError('manifest identity mismatch') + if harness not in ('claude','codex') or not isinstance(m['source_sha256'],str) or not HEX.fullmatch(m['source_sha256']): + raise ValueError('manifest identity/digest invalid') + if m['previous'] is not None: + iri(m['previous']) + if not isinstance(m['parts'],list): + raise ValueError('invalid parts') + total = 0 + for i,p in enumerate(m['parts']): + if not isinstance(p,list) or len(p)!=4 or type(p[0]) is not int or p[0]!=i: + raise ValueError('duplicate/noncontiguous partition') + if not isinstance(p[1],str) or not HEX.fullmatch(p[1]) or not 1<=count(p[2])<=PART_RECORDS or not 1<=count(p[3])<=PART_BYTES: + raise ValueError('part budget/digest invalid') + total += p[2] + if count(m['records']) != total: + raise ValueError('manifest count mismatch') + return m + + +def prepare_rows(records): + """One session source, unique request keys; no implicit cross-harness merge.""" + if not isinstance(records, list): + raise ValueError('records must be an array') + rows = sorted((record_row(r) for r in records), key=lambda r:r[0]) + keys = [tuple(r[0]) for r in rows] + if len(keys) != len(set(keys)): + raise ValueError('duplicate request identity') + if len({tuple(r[0][:2]) for r in rows}) > 1 or len({r[-1] for r in rows}) > 1: + raise ValueError('mixed session/harness/actor source') + return rows diff --git a/scripts/usage_generation_publisher.py b/scripts/usage_generation_publisher.py new file mode 100644 index 0000000..2c54693 --- /dev/null +++ b/scripts/usage_generation_publisher.py @@ -0,0 +1,176 @@ +"""Explicit opt-in staged generation ticks, sharing the existing cost cadence.""" +from datetime import datetime,timezone +import hashlib +import json +import time +from pathlib import Path + +import planes +import usage_generation_codec as c +from usage_generation_wire import plan,byte_size +from usage_generation_budget import admit_session +from usage_generation_step import atomic,locked,run_step +from usage_generation_terms import parse_response,snapshot_query +from usage_generation_reader import read_snapshot + +PROFILE='camayoc-usage-generations' +PROFILE_FILE=Path(__file__).resolve().parents[1]/'shapes/usage-generations.shapes.ttl' + + +def policy_check(policy): + required={'enabled','producer','harness','session','legacy_frozen','runtime_acceptance'} + if not isinstance(policy,dict) or set(policy)!=required: + raise ValueError('generation policy schema missing/unknown') + if policy['enabled'] is not True or policy['legacy_frozen'] is not True: + raise ValueError('generation path not enrolled/legacy publication not frozen') + c.iri(policy['producer']);c.text(policy['session']);c.text(policy['runtime_acceptance']) + if policy['harness'] not in ('claude','codex'):raise ValueError('unknown generation harness') + + +def load_state(path): + if not path.exists():return {'version':1,'sessions':{},'profile':None} + value=json.loads(path.read_text()) + if not isinstance(value,dict) or set(value)!={'version','sessions','profile'} or value['version']!=1 or not isinstance(value['sessions'],dict): + raise ValueError('UNKNOWN: invalid generation journal') + return value + + +def query_triples(graph,subjects): + values=' '.join('<'+c.iri(s)+'>' for s in sorted(subjects)) + return {'query':'SELECT ?s ?p (STR(?o) AS ?lex) (isIRI(?o) AS ?iri) ' + '(DATATYPE(?o) AS ?datatype) (LANG(?o) AS ?lang) WHERE { ' + 'GRAPH <'+c.iri(graph)+'> { VALUES ?s { '+values+' } ?s ?p ?o } } LIMIT 10000', + 'verbose':True} + + +def verify_triples(call,graph,triples): + expected={(graph,s,p,tuple(o)) for s,p,o in triples} + actual=set(parse_response(call('/query',query_triples(graph,{s for _,s,_,_ in expected})),graph)) + return actual==expected + + +def tick(result,actor,state_path,receipt,policy,*,lane_locked=False): + policy_check(policy) + if result.get('errors'):raise ValueError('incomplete parser input') + state_path=Path(state_path);journal_path=state_path.with_suffix('.generations.json') + # Separate planner lock plus shared write-step lock; C uses the latter. + with locked(state_path.with_name(state_path.name+'.planner')): + if state_path.with_suffix('.pending.json').exists(): + raise ValueError('INDETERMINATE: reconcile pending write') + state=load_state(journal_path);now=receipt['attempted_at'] + expected_profile=hashlib.sha256(PROFILE_FILE.read_bytes()).hexdigest() + def post(endpoint,body): + if receipt['requests']>=3:raise ValueError('cost request budget') + receipt.setdefault('first_request_at',time.time()) + receipt['requests']+=1 + started=time.monotonic() + try:return planes._post(endpoint,body,client='camayoc-cost-generation',response_limit=c.READ_BYTES if endpoint=='/query' else 262144) + finally:receipt['request_seconds']+=time.monotonic()-started + if state['profile']!=expected_profile: + loaded=post('/shapes',{'action':'get','name':PROFILE}) + if not isinstance(loaded.get('turtle'),str) or hashlib.sha256(loaded['turtle'].encode()).hexdigest()!=expected_profile: + raise ValueError('UNKNOWN: exact governed profile not loaded') + state['profile']=expected_profile;atomic(journal_path,state) + receipt.update(generation_phase='profile-verified',unknown_reason='staging') + return [] + key=c.digest([policy['producer'],policy['harness'],policy['session']]) + entry=state['sessions'].get(key,{'selected':None,'progress':None,'retained':[]}) + if set(entry)!={'selected','progress','retained'} or not isinstance(entry['retained'],list): + raise ValueError('UNKNOWN: invalid session journal') + source_rows=c.prepare_rows(result['records']) + if source_rows and source_rows[0][0][:2]!=[policy['harness'],policy['session']]: + raise ValueError('mixed generation source identity') + source=c.digest(source_rows) + progress=entry['progress'] + if progress is None: + selected=entry['selected'] + if selected and selected['source']==source: + receipt.update(generation_phase='unchanged-selected',readback='sealed-generation') + return [] + if entry['retained']: + raise ValueError('UNKNOWN: retained graph occupies generation slot; governed retention required') + items={r['bead'] for r in result['records'] if r['attribution']=='attributed'} + # Corrections stay on A after selection, including an explicitly + # complete empty input; never resume the old whole-session writer. + if not selected:admit_session(c_items=len(items),source_digest=source,previous={},staged_bytes=0,body_bytes=0) + horizon=datetime.fromtimestamp(now,timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') + planned=plan(result['records'],producer=policy['producer'],harness=policy['harness'], + session=policy['session'],at=horizon,actor=actor, + previous=selected['generation'] if selected else None) + total=sum(e['progress']['plan']['bytes'] for e in state['sessions'].values() if e.get('progress')) + if total+planned['bytes']>c.STAGED_BYTES:raise ValueError('aggregate staged bytes exceed64MiB') + progress={'plan':planned,'phase':'control','part':0,'receipts':[]} + entry['progress']=progress;state['sessions'][key]=entry;atomic(journal_path,state) + planned=progress['plan'];phase=progress['phase'];index=progress['part'] + if phase in ('control','graph'): + graph=planned['control'] if phase=='control' else planned['generation'] + # Graph creation has its OWN tick and occupies the shared write lane. + body={'graph':graph};kind='graph';n=0;preflight=None + def verify(call,accepted): + # Exact graph registry identity from the native receipt. A + # graph-bound query must also answer, including valid empty. + answer=call('/query',{'query':'SELECT ?s WHERE { GRAPH <'+graph+'> { ?s ?p ?o } } LIMIT 1'}) + return accepted.get('graph',graph)==graph and isinstance(answer.get('rows'),list) and answer.get('truncated') is False + elif phase in ('control-label','graph-label'): + graph=planned['control'] if phase=='control-label' else planned['generation'] + body={'graph':graph,'timestamp':planned['manifest']['horizon'],'kind':'knowledge','actor':actor} + kind='graph_label';n=0;preflight=None + def verify(call,_): + answer=call('/query',{'graph':graph,'query':'SELECT ?s WHERE { ?s ?p ?o } LIMIT 1'}) + return answer.get('truncated') is False and answer.get('labels',{}).get('kind')=={'coverage':'full','value':['knowledge']} + elif phase=='part': + body=planned['parts'][index];kind='part';n=planned['manifest']['parts'][index][2] + triples=planned['part_triples'][index] + items=sorted({o[1] for _,p,o in triples if p==c.ONTOLOGY+'attributedTo'}) + def preflight(call): + if not items:return True + values=' '.join('<'+c.iri(s)+'>' for s in items) + answer=call('/query',{'graph':planes.plane_for('observed'), + 'query':'SELECT ?item WHERE { VALUES ?item { '+values+' } ?item a ?t . FILTER(?t = <'+c.ONTOLOGY+'WorkItem>) }','verbose':True}) + return answer.get('truncated') is False and {r.get('item') for r in answer.get('rows',[])}==set(items) + verify=lambda call,_:verify_triples(call,planned['generation'],triples) + elif phase=='manifest': + body=planned['metadata'];kind='part';n=0;preflight=None + verify=lambda call,_:verify_triples(call,planned['generation'],planned['metadata_triples']) + elif phase=='activate': + body=planned['activation'];kind='marker';n=0;preflight=None + def verify(call,_): + checked=read_snapshot(call('/query',snapshot_query(planned['control'],planned['selector'])), + producer=policy['producer'],harness=policy['harness'],session=policy['session'],at=planned['manifest']['horizon']) + return checked['digest']==c.digest(planned['manifest']) + else:raise ValueError('UNKNOWN: invalid generation phase') + # Recover cursor-only crash from the durable VERIFIED exact-body proof, + # never replay an already verified write. + old_path=state_path.with_suffix('.generation-receipt.json') + prior=json.loads(old_path.read_text()) if old_path.exists() else {} + receipt['items_per_snapshot'].append(len({r[8][0] for r in planned['rows'] if r[8]})) + receipt['records_per_snapshot'].append(n) + receipt['body_bytes_per_snapshot'].append(byte_size(body)) + body_hash=hashlib.sha256(json.dumps(body,sort_keys=True).encode()).hexdigest() + recovered=prior.get('status')=='VERIFIED' and prior.get('body_sha256')==body_hash + if recovered: + accepted=prior + else: + accepted=run_step(kind,body,state_path,now=now,post=post,verify=verify, + preflight=preflight,records=n,lane_locked=lane_locked,staged_bytes=sum(e['progress']['plan']['bytes'] for e in state['sessions'].values() if e.get('progress'))) + progress['receipts'].append({'phase':phase,'index':index,'sha256':body_hash, + 'bytes':byte_size(body),'accepted':accepted.get('accepted')}) + if phase=='control':progress['phase']='control-label' + elif phase=='control-label':progress['phase']='graph' + elif phase=='graph':progress['phase']='graph-label' + elif phase=='graph-label':progress['phase']='part' if planned['parts'] else 'manifest' + elif phase=='part': + progress['part']+=1 + if progress['part']==len(planned['parts']):progress['phase']='manifest' + elif phase=='manifest':progress['phase']='activate' + else: + if entry['selected']:entry['retained'].append(entry['selected']) + entry['selected']={'generation':planned['generation'],'source':planned['manifest']['source_sha256'], + 'producer':policy['producer'],'harness':policy['harness'],'session':policy['session'], + 'horizon':planned['manifest']['horizon'],'bytes':planned['bytes']} + entry['progress']=None + atomic(journal_path,state) + receipt.update(generation_phase=phase,readback='exact-generation' if phase=='activate' else 'exact-step', + recovered_cursor=recovered,knot_seconds=0 if recovered else accepted.get('knot_seconds',0), + unknown_reason='none' if phase=='activate' else 'staging') + return [accepted.get('accepted',{}).get('tx_id')] if not recovered and phase not in ('control','graph','control-label','graph-label') else [] diff --git a/scripts/usage_generation_reader.py b/scripts/usage_generation_reader.py new file mode 100644 index 0000000..14c1891 --- /dev/null +++ b/scripts/usage_generation_reader.py @@ -0,0 +1,141 @@ +"""One-response complete-set validation. Errors are UNKNOWN, never partial cost.""" +import re +from urllib.parse import quote,unquote + +import usage_generation_codec as c +from usage_generation_terms import parse_response +from usage_generation_wire import RDF,RDFS,XSD,U,S,I + + +class Facts: + def __init__(self,triples): + self.values={};self.used=set() + for s,p,o in triples:self.values.setdefault((s,p),[]).append(o) + + def one(self,s,p,kind='literal',datatype=XSD+'string',optional=False): + key=(s,p);values=self.values.get(key,[]) + if optional and not values:return None + if len(values)!=1:raise ValueError('UNKNOWN: scalar cardinality') + term=values[0] + if term[0]!=kind or term[2]!=(None if kind=='uri' else datatype): + raise ValueError('UNKNOWN: wrong scalar RDF kind/datatype') + self.used.add(key);return term[1] + + def exact(self,s,p,value): + got=self.one(s,p,value[0],value[2]) + if got!=value[1]:raise ValueError('UNKNOWN: fixed fact differs') + + def done(self): + if set(self.values)!=self.used:raise ValueError('UNKNOWN: unexpected facts') + + +def read_snapshot(response,*,producer,harness,session,at): + control=c.BASE+'selector-graph/'+c.digest([producer,harness,session]) + selector=c.BASE+'selector/'+c.digest([producer,harness,session]) + quads=parse_response(response,control) + cf=Facts([(s,p,o) for g,s,p,o in quads if g==control]);a=c.ONTOLOGY + cf.exact(selector,RDF+'type',U(a+'UsageGenerationSelector')) + for p,t in [('inSession',U(c.BASE+'session/'+quote(session,safe=''))), + ('usageHarness',S(harness)),('usageProducer',U(producer)), + ('usageProfileVersion',S(c.VERSION)),('usageHorizon',S(c.horizon(at)))]: + cf.exact(selector,a+p,t) + gen=cf.one(selector,a+'usageSelectedGeneration','uri') + selected_digest=cf.one(selector,a+'usageManifestDigest');cf.done() + if not c.HEX.fullmatch(selected_digest) or gen!=c.BASE+'generation/'+selected_digest: + raise ValueError('UNKNOWN: selector digest/identity mismatch') + if {g for g,*_ in quads}!={control,gen}: + raise ValueError('UNKNOWN: mixed/unselected/legacy graph') + gf=Facts([(s,p,o) for g,s,p,o in quads if g==gen]) + gf.exact(gen,RDF+'type',U(a+'UsageGeneration')) + for p,t in [('inSession',U(c.BASE+'session/'+quote(session,safe=''))), + ('usageHarness',S(harness)),('usageProducer',U(producer)), + ('usageProfileVersion',S(c.VERSION)),('usageHorizon',S(at)),('usageGraph',U(gen))]: + gf.exact(gen,a+p,t) + raw=gf.one(gen,a+'usageManifestJSON') + m=c.validate_manifest(raw,producer=producer,harness=harness,session=session,at=at) + if c.digest(m)!=selected_digest:raise ValueError('UNKNOWN: manifest digest differs') + gf.exact(gen,a+'usageManifestDigest',S(selected_digest)) + gf.exact(gen,a+'usageSourceDigest',S(m['source_sha256'])) + gf.exact(gen,a+'usagePartitionCount',I(len(m['parts']))) + gf.exact(gen,a+'usageRecordCount',I(m['records'])) + previous=gf.one(gen,a+'usagePreviousGeneration','uri',optional=True) + if previous!=m['previous']:raise ValueError('UNKNOWN: correction lineage mismatch') + subjects={s for s,p in gf.values if p==RDF+'type' and U(a+'UsageRecord') in gf.values[(s,p)]} + rows=[];sessions=set();items=set() + for s in subjects: + gf.exact(s,RDF+'type',U(a+'UsageRecord'));gf.exact(s,a+'sourceKind',S('observed')) + gf.one(s,RDFS+'label') + si=gf.one(s,a+'inSession','uri') + if si!=c.BASE+'session/'+quote(session,safe=''):raise ValueError('UNKNOWN: mixed session') + sessions.add(si);gf.exact(s,a+'usageHarness',S(harness)) + request=gf.one(s,a+'usageRequestId');c.text(request) + provider=gf.one(s,a+'provider') + if provider!=('openai' if harness=='codex' else 'anthropic'):raise ValueError('UNKNOWN: provider') + def number(p,optional=False): + v=gf.one(s,a+p,datatype=XSD+'integer',optional=optional) + if v is not None and not re.fullmatch(r'0|[1-9][0-9]*',v):raise ValueError('UNKNOWN: noncanonical count') + return v + tokens=number('tokensConsumed') + counts=[number(p,True) for p in ('inputTokensUncached','cacheReadInputTokens','cacheWriteInputTokens','outputTokens')] + model=gf.one(s,a+'usageModel',optional=True);ms=gf.one(s,a+'usageModelSource',optional=True) + stamp=gf.one(s,a+'observedAt');state=gf.one(s,a+'usageAttributionState') + item=gf.one(s,a+'attributedTo','uri',optional=True) + if (state not in ('attributed','no proven focus','overlapping focus') + or (state=='attributed')!=(item is not None)): + raise ValueError('UNKNOWN: contradictory attribution') + if item:items.add(item) + actor_iri=gf.one(si,a+'actor','uri') + if not actor_iri.startswith(c.BASE+'principal/'):raise ValueError('UNKNOWN: actor route') + actor=unquote(actor_iri.removeprefix(c.BASE+'principal/')) + if c.BASE+'principal/'+quote(actor,safe='')!=actor_iri:raise ValueError('UNKNOWN: actor encoding') + row=[[harness,session,request],tokens,counts,provider,model,ms,stamp,state,[item] if item else [],actor] + if s!=c.record_iri(row):raise ValueError('UNKNOWN: request identity differs') + rows.append(row) + for si in sessions: + gf.exact(si,RDF+'type',U(a+'Session'));gf.exact(si,a+'sourceKind',S('observed')) + gf.exact(si,RDFS+'label',S(harness+' session '+session)) + for item in items: + if not item.startswith(a):raise ValueError('UNKNOWN: WorkItem namespace') + gf.exact(item,RDF+'type',U(a+'WorkItem')) + gf.exact(item,RDFS+'label',S(item.removeprefix(a))) + gf.done();rows.sort(key=lambda r:r[0]) + if len(rows)!=m['records'] or len({tuple(r[0]) for r in rows})!=len(rows) or c.digest(rows)!=m['source_sha256']: + raise ValueError('UNKNOWN: record set/source digest differs') + index=0 + for _,sha,n,_bytes in m['parts']: + if c.digest(rows[index:index+n])!=sha:raise ValueError('UNKNOWN: missing/tampered/mixed partition') + index+=n + return {'status':'COMPLETE','manifest':m,'rows':rows,'generation':gen,'digest':selected_digest, + 'scope':'one supplied session snapshot; source-log/global completeness not asserted'} + + +def consume(*,producer,harness,session,at,state_path,now,post=None): + """Read-only consumer build, one graph-bound response per5min; no polling.""" + import json + from pathlib import Path + import planes + from usage_generation_budget import admit_read + from usage_generation_step import locked,atomic + from usage_generation_terms import snapshot_query + c.iri(producer);c.text(session);c.horizon(at) + if harness not in ('claude','codex'):raise ValueError('UNKNOWN: invalid reader harness') + state_path=Path(state_path) + control=c.BASE+'selector-graph/'+c.digest([producer,harness,session]) + selector=c.BASE+'selector/'+c.digest([producer,harness,session]) + transport=post or planes._post + with locked(state_path): + state=json.loads(state_path.read_text()) if state_path.exists() else {'attempt':None,'complete':None} + if not isinstance(state,dict) or set(state)!={'attempt','complete'}: + raise ValueError('UNKNOWN: invalid consumer state') + try:attempt=admit_read(now=now,previous=state['attempt'],graph=control) + except ValueError: + return {'status':'UNKNOWN','reason':'consumer_cadence_or_clock','last_complete':state['complete']} + # Failed attempts count too; never poll after an error or lower the cap. + state['attempt']=attempt;atomic(state_path,state) + try: + response=transport('/query',snapshot_query(control,selector),client='camayoc-cost-generation',response_limit=c.READ_BYTES) + checked=read_snapshot(response,producer=producer,harness=harness,session=session,at=at) + except (ValueError,OSError,planes.PlaneError): + return {'status':'UNKNOWN','reason':'generation_validation','last_complete':state['complete']} + state['complete']=checked;atomic(state_path,state) + return checked diff --git a/scripts/usage_generation_step.py b/scripts/usage_generation_step.py new file mode 100644 index 0000000..0bc3c33 --- /dev/null +++ b/scripts/usage_generation_step.py @@ -0,0 +1,113 @@ +"""One bounded, pending-before-send cost step; no timer or implicit enrollment. + +Only a reviewed caller holding the cost producer lane may invoke this primitive. +The wire body and verification remain domain-owned. An uncertain write is never +retried here. The shared journal counts C and A attempts, including failures. +""" +from contextlib import contextmanager, nullcontext +import fcntl +import hashlib +import json +import os +from pathlib import Path +import time + +from usage_generation_budget import admit, MAX_REQUESTS + + +def atomic(path, value): + path = Path(path) + temporary = path.with_name(path.name+'.tmp') + with temporary.open('w') as out: + json.dump(value,out,sort_keys=True,separators=(',',':')) + out.flush(); os.fsync(out.fileno()) + os.replace(temporary,path) + fd=os.open(path.parent,os.O_RDONLY) + try: os.fsync(fd) + finally: os.close(fd) + + +@contextmanager +def locked(state): + path=Path(state).with_suffix('.generations.lock') + with path.open('a') as handle: + try: fcntl.flock(handle,fcntl.LOCK_EX|fcntl.LOCK_NB) + except BlockingIOError: raise ValueError('cost producer lane busy') from None + try: yield + finally: fcntl.flock(handle,fcntl.LOCK_UN) + + +def run_step(kind, body, state_path, *, now, post, verify, preflight=None, + records=0, staged_bytes=0, clock=time.monotonic, lane_locked=False): + """One write, exact-body pending fence, bounded queries and conservative rate. + + post(endpoint, body) is the existing reviewed transport, never an alternate + store or credential resolver. verify(call, accepted) must establish exact + reader-route proof and may issue at most one query. Success clears pending + only after the verified receipt is durably written. No cleanup on error. + """ + state_path=Path(state_path) + pending=state_path.with_suffix('.pending.json') + ledger_path=state_path.with_suffix('.writes.json') + receipt_path=state_path.with_suffix('.generation-receipt.json') + with (nullcontext() if lane_locked else locked(state_path)): + if pending.exists(): + raise ValueError('INDETERMINATE: reconcile pending write before any resend') + previous=json.loads(receipt_path.read_text()) if receipt_path.exists() else {} + if not isinstance(previous,dict) or previous.get('next_request_after',0)>now: + raise ValueError('BACKOFF: previous request budget cooldown') + ledger=json.loads(ledger_path.read_text()) if ledger_path.exists() else [] + encoded=json.dumps(body,sort_keys=True).encode() + maximum=MAX_REQUESTS.get(kind) + if maximum is None or preflight is not None and kind!='part': + raise ValueError('invalid cost action/preflight') + proposed=admit(kind,len(encoded),records,maximum,now,ledger,staged_bytes) + receipt={'at':now,'status':'UNKNOWN','requests':0,'write_sent':False, + 'body_sha256':hashlib.sha256(encoded).hexdigest(),'bytes':len(encoded), + 'kind':kind,'request_seconds':0,'knot_seconds':0} + last=None + def call(endpoint, payload): + nonlocal last + if endpoint!='/query' or receipt['requests']>=maximum: + raise ValueError('only the bounded query slot is available') + if last is not None: time.sleep(max(0,1-(clock()-last))) + last=clock();receipt['requests']+=1;started=clock() + try: return post(endpoint,payload) + finally: receipt['request_seconds']+=clock()-started + try: + if preflight is not None and preflight(call) is not True: + raise ValueError('canonical preflight unproven') + if receipt['requests']>=maximum: + raise ValueError('request budget exhausted before write') + # Persist the body and attempt budget BEFORE networking. A crash can + # conservatively consume a slot; it must never enable a resend. + atomic(pending,{'endpoint':('/graph/create' if kind=='graph' else '/graph/label' if kind=='graph_label' else '/knot'), + 'body':body,'receipt':dict(receipt)}) + atomic(ledger_path,proposed) + if last is not None: time.sleep(max(0,1-(clock()-last))) + last=clock();receipt['requests']+=1;receipt['write_sent']=True;started=clock() + try: accepted=post('/graph/create' if kind=='graph' else '/graph/label' if kind=='graph_label' else '/knot',body) + finally: + elapsed=clock()-started;receipt['knot_seconds']=elapsed + receipt['request_seconds']+=elapsed + if not isinstance(accepted,dict) or accepted.get('conforms') is False: + raise ValueError('write refused/unknown') + if kind=='graph': + if type(accepted.get('g')) is not int: + raise ValueError('graph registration receipt missing') + elif type(accepted.get('tx_id')) is not int: + raise ValueError('write transaction receipt missing') + verification_start=receipt['requests'] + if verify(call,accepted) is not True or receipt['requests']-verification_start!=1: + raise ValueError('one exact graph readback unproven') + receipt.update(status='VERIFIED',accepted=accepted) + slow=receipt['knot_seconds']>10 or receipt['request_seconds']>15 + receipt['next_request_after']=now+(900 if slow else 60) + atomic(receipt_path,receipt) + pending.unlink() + return receipt + finally: + slow=receipt['knot_seconds']>10 or receipt['request_seconds']>15 + receipt['next_request_after']=now+(900 if slow else 60) + if pending.exists(): receipt['status']='STALLED' + atomic(receipt_path,receipt) diff --git a/scripts/usage_generation_terms.py b/scripts/usage_generation_terms.py new file mode 100644 index 0000000..feaea3b --- /dev/null +++ b/scripts/usage_generation_terms.py @@ -0,0 +1,56 @@ +"""Lossless-term projection for a single bounded generation query. + +STR/isIRI/DATATYPE/LANG are explicit in the one normal JSON response, retaining +its truncation flag; standard SPARQL JSON silently drops that server flag. +No selected generation or completeness verdict is cached across requests here. +""" +from usage_generation_codec import iri, ONTOLOGY + +MAX_ROWS = 10000 +XSD = 'http://www.w3.org/2001/XMLSchema#' + + +def snapshot_query(control_graph, selector): + control=iri(control_graph); subject=iri(selector) + return {'query': ( + 'SELECT ?g ?s ?p (STR(?o) AS ?lex) (isIRI(?o) AS ?iri) ' + '(DATATYPE(?o) AS ?datatype) (LANG(?o) AS ?lang) WHERE { ' + '{ GRAPH <'+control+'> { ?s ?p ?o } } UNION { ' + 'GRAPH <'+control+'> { <'+subject+'> <'+ONTOLOGY+'usageSelectedGeneration> ?g } ' + 'GRAPH ?g { ?s ?p ?o } } } LIMIT '+str(MAX_ROWS)), 'verbose':True} + + +def parse_response(response, control_graph): + """Return deduplicated graph/subject/predicate/term tuples, or UNKNOWN error.""" + if (not isinstance(response,dict) or response.get('truncated') is not False + or not isinstance(response.get('rows'),list) + or type(response.get('count')) is not int + or response['count']!=len(response['rows']) + or len(response['rows'])>=MAX_ROWS or not response['rows']): + raise ValueError('UNKNOWN: empty/truncated/noncanonical generation response') + control=iri(control_graph); result=[];seen=set() + for row in response['rows']: + if not isinstance(row,dict) or set(row)-{'g','s','p','lex','iri','datatype','lang'}: + raise ValueError('UNKNOWN: malformed term row') + graph=iri(row['g']) if 'g' in row else control + subject=iri(row.get('s'));predicate=iri(row.get('p')) + lexical=row.get('lex') + if not isinstance(lexical,str) or type(row.get('iri')) is not bool: + raise ValueError('UNKNOWN: missing term kind/lexical value') + lexical.encode('utf-8','strict') + if row['iri']: + if 'datatype' in row or 'lang' in row: + raise ValueError('UNKNOWN: contradictory IRI metadata') + term=('uri',iri(lexical),None) + else: + datatype=iri(row.get('datatype')) + if row.get('lang','')!='': + raise ValueError('UNKNOWN: language-tagged accounting term') + if datatype not in (XSD+'string',XSD+'integer'): + raise ValueError('UNKNOWN: unsupported accounting literal type') + term=('literal',lexical,datatype) + triple=(graph,subject,predicate,term) + if triple in seen: + raise ValueError('UNKNOWN: duplicate response tuple') + seen.add(triple);result.append(triple) + return result diff --git a/scripts/usage_generation_wire.py b/scripts/usage_generation_wire.py new file mode 100644 index 0000000..06b8e9d --- /dev/null +++ b/scripts/usage_generation_wire.py @@ -0,0 +1,126 @@ +"""Exact proposal42/43 RDF and bounded per-session generation plans.""" +import json +from urllib.parse import quote + +import usage_generation_codec as c + +RDF='http://www.w3.org/1999/02/22-rdf-syntax-ns#' +RDFS='http://www.w3.org/2000/01/rdf-schema#' +XSD='http://www.w3.org/2001/XMLSchema#' + + +def U(value): return ('uri',c.iri(value),None) +def S(value): return ('literal',c.text(value),XSD+'string') +def I(value): return ('literal',str(c.count(value)),XSD+'integer') + + +def turtle(triples): + def uri(value): + value=c.iri(value) + if value.startswith(c.ONTOLOGY) and value[len(c.ONTOLOGY):].isalnum(): + return 'a:'+value[len(c.ONTOLOGY):] + return '<'+value+'>' + def obj(t): + return uri(t[1]) if t[0]=='uri' else json.dumps(t[1],ensure_ascii=False)+'^^'+uri(t[2]) + subjects={} + for subject,predicate,term in sorted(set(triples)): + subjects.setdefault(subject,[]).append((predicate,term)) + lines=['@prefix a: <'+c.ONTOLOGY+'> .'] + for subject,properties in subjects.items(): + lines.append(uri(subject)+' '+' ; '.join((('a' if p==RDF+'type' else uri(p))+' '+obj(t)) for p,t in properties)+' .') + return '\n'.join(lines) + + +def session_iri(row): return c.BASE+'session/'+quote(row[0][1],safe='') + + +def record_triples(row): + key,tokens,counts,provider,model,model_source,stamp,state,items,actor=row + subject=c.record_iri(row);session=session_iri(row) + a=c.ONTOLOGY + out=[(session,RDF+'type',U(a+'Session')),(session,RDFS+'label',S(key[0]+' session '+key[1])), + (session,a+'sourceKind',S('observed')),(session,a+'actor',U(c.BASE+'principal/'+quote(actor,safe=''))), + (subject,RDF+'type',U(a+'UsageRecord')),(subject,RDFS+'label',S(key[2])), + (subject,a+'sourceKind',S('observed')),(subject,a+'inSession',U(session)), + (subject,a+'provider',S(provider)),(subject,a+'tokensConsumed',I(int(tokens))), + (subject,a+'observedAt',S(stamp)),(subject,a+'usageHarness',S(key[0])), + (subject,a+'usageRequestId',S(key[2])),(subject,a+'usageAttributionState',S(state))] + for p,v in [('usageModel',model),('usageModelSource',model_source)]: + if v is not None: out.append((subject,a+p,S(v))) + for p,v in zip(('inputTokensUncached','cacheReadInputTokens','cacheWriteInputTokens','outputTokens'),counts): + if v is not None: out.append((subject,a+p,I(int(v)))) + for item in items: + out.append((subject,a+'attributedTo',U(item))) + # Graph-local reference stubs are NOT canonical tracker writes. The + # direct type must be proved in the observed graph before staging. + # The label states the identifier only, never a guessed tracker title. + out.extend([(item,RDF+'type',U(a+'WorkItem')),(item,RDFS+'label',S(item.removeprefix(a)))]) + return out + + +def common(producer,harness,session,at): + return [('inSession',U(c.BASE+'session/'+quote(session,safe=''))), + ('usageHarness',S(harness)),('usageProducer',U(producer)), + ('usageProfileVersion',S(c.VERSION)),('usageHorizon',S(c.horizon(at)))] + + +def knot(graph,snapshot,triples,actor): + return {'graph':c.iri(graph),'snapshot':snapshot,'replace_snapshot':True, + 'actor':c.text(actor),'source':'camayoc-cost-generation','turtle':turtle(triples)} + + +def byte_size(body): return len(json.dumps(body,sort_keys=True).encode()) + + +def plan(records,*,producer,harness,session,at,actor,previous=None): + rows=c.prepare_rows(records);c.iri(producer);c.text(session);c.horizon(at) + from work_cost import epoch + if any(epoch(row[6])>epoch(at) for row in rows):raise ValueError('source record beyond captured horizon') + if harness not in ('claude','codex') or rows and rows[0][0][:2]!=[harness,session]: + raise ValueError('session scope mismatch') + if previous is not None:c.iri(previous) + placeholder=c.BASE+'generation/'+'0'*64 + chunks=[];current=[] + def body(i,chunk,graph): + triples=[t for row in chunk for t in record_triples(row)] + return knot(graph,'camayoc-cost:'+graph+':'+str(i),triples,actor) + for row in rows: + trial=current+[row] + if len(trial)>c.PART_RECORDS or byte_size(body(len(chunks),trial,placeholder))>c.PART_BYTES: + if not current:raise ValueError('single request exceeds part budget') + chunks.append(current);current=[row] + if byte_size(body(len(chunks),current,placeholder))>c.PART_BYTES: + raise ValueError('single request exceeds part budget') + else:current=trial + if current:chunks.append(current) + descriptors=[[i,c.digest(chunk),len(chunk),byte_size(body(i,chunk,placeholder))] for i,chunk in enumerate(chunks)] + manifest={'version':c.VERSION,'producer':producer,'harness':harness,'session':session, + 'horizon':at,'source_sha256':c.digest(rows),'previous':previous, + 'parts':descriptors,'records':len(rows)} + raw=c.canonical(manifest);md=c.digest(manifest);generation=c.BASE+'generation/'+md + control=c.BASE+'selector-graph/'+c.digest([producer,harness,session]) + selector=c.BASE+'selector/'+c.digest([producer,harness,session]) + parts=[body(i,chunk,generation) for i,chunk in enumerate(chunks)] + if [byte_size(b) for b in parts]!=[p[3] for p in descriptors]:raise ValueError('body size identity changed') + props=common(producer,harness,session,at) + envelope=[(generation,RDF+'type',U(c.ONTOLOGY+'UsageGeneration'))] + envelope.extend((generation,c.ONTOLOGY+p,v) for p,v in props) + envelope.extend((generation,c.ONTOLOGY+p,v) for p,v in [ + ('usageGraph',U(generation)),('usageSourceDigest',S(manifest['source_sha256'])), + ('usageManifestDigest',S(md)),('usageManifestJSON',S(raw)), + ('usagePartitionCount',I(len(chunks))),('usageRecordCount',I(len(rows)))]) + if previous:envelope.append((generation,c.ONTOLOGY+'usagePreviousGeneration',U(previous))) + metadata=knot(generation,'camayoc-cost:'+generation+':manifest',envelope,actor) + marker=[(selector,RDF+'type',U(c.ONTOLOGY+'UsageGenerationSelector'))] + marker.extend((selector,c.ONTOLOGY+p,v) for p,v in props) + marker.extend([(selector,c.ONTOLOGY+'usageSelectedGeneration',U(generation)), + (selector,c.ONTOLOGY+'usageManifestDigest',S(md))]) + activation=knot(control,'camayoc-cost:selector:'+selector,marker,actor) + if byte_size(metadata)>c.PART_BYTES or byte_size(activation)>c.MARKER_BYTES: + raise ValueError('manifest/marker body budget') + total=sum(byte_size(b) for b in parts)+byte_size(metadata) + if total>c.STAGED_BYTES:raise ValueError('generation staged-byte budget') + return {'manifest':manifest,'generation':generation,'control':control,'selector':selector, + 'parts':parts,'metadata':metadata,'activation':activation,'rows':rows,'bytes':total, + 'part_triples':[[t for row in chunk for t in record_triples(row)] for chunk in chunks], + 'metadata_triples':envelope,'activation_triples':marker} diff --git a/shapes/usage-generations.shapes.ttl b/shapes/usage-generations.shapes.ttl new file mode 100644 index 0000000..45a02a8 --- /dev/null +++ b/shapes/usage-generations.shapes.ttl @@ -0,0 +1,47 @@ +# PROPOSAL ONLY: aegis-srm; owner Camayoc. Not authorized/loaded in production. +@prefix a: . +@prefix sh: . +@prefix xsd: . +@prefix rdf: . +@prefix rdfs: . +a:UsageGeneration a rdfs:Class ; rdfs:comment "PROPOSED Camayoc usage envelope; not a UsageRecord and not a completeness assertion." . +a:UsageGenerationSelector a rdfs:Class ; rdfs:comment "PROPOSED Camayoc usage envelope; not a UsageRecord and not a completeness assertion." . + +a:UsageGenerationShape a sh:NodeShape ; sh:targetClass a:UsageGeneration ; + sh:property [ sh:path a:inSession ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ; + sh:property [ sh:path a:usageHarness ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:in ("claude" "codex") ] ; + sh:property [ sh:path a:usageProducer ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ; + sh:property [ sh:path a:usageProfileVersion ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:in ("srm-generation-v1") ] ; + sh:property [ sh:path a:usageHorizon ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:pattern "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$" ] ; + sh:property [ sh:path a:usageGraph ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ; + sh:property [ sh:path a:usageSourceDigest ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:pattern "^[0-9a-f]{64}$" ] ; + sh:property [ sh:path a:usageManifestDigest ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:pattern "^[0-9a-f]{64}$" ] ; + sh:property [ sh:path a:usageManifestJSON ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ] ; + sh:property [ sh:path a:usagePartitionCount ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:integer ; sh:minInclusive 0 ; sh:lessThanOrEquals a:usageRecordCount ] ; + sh:property [ sh:path a:usageRecordCount ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:integer ; sh:minInclusive 0 ] ; + sh:property [ sh:path a:usagePreviousGeneration ; sh:minCount 0 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] . + +a:UsageGenerationSelectorShape a sh:NodeShape ; sh:targetClass a:UsageGenerationSelector ; + sh:property [ sh:path a:inSession ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ; + sh:property [ sh:path a:usageHarness ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:in ("claude" "codex") ] ; + sh:property [ sh:path a:usageProducer ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ; + sh:property [ sh:path a:usageProfileVersion ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:in ("srm-generation-v1") ] ; + sh:property [ sh:path a:usageHorizon ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:pattern "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$" ] ; + sh:property [ sh:path a:usageSelectedGeneration ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ; + sh:property [ sh:path a:usageManifestDigest ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:pattern "^[0-9a-f]{64}$" ] . + +a:UsageRequestIdentityShape a sh:NodeShape ; sh:targetSubjectsOf a:usageRequestId ; + sh:property [ sh:path a:usageRequestId ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:minLength 1 ] ; + sh:property [ sh:path a:inSession ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ; + sh:property [ sh:path a:usageHarness ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:in ("claude" "codex") ] ; + sh:property [ sh:path a:usageAttributionState ; sh:minCount 1 ; sh:maxCount 1 ; sh:datatype xsd:string ; sh:in ("attributed" "no proven focus" "overlapping focus") ] ; + sh:or ( + [ sh:and ( + [ sh:property [ sh:path a:usageAttributionState ; sh:hasValue "attributed" ] ] + [ sh:property [ sh:path a:attributedTo ; sh:minCount 1 ; sh:maxCount 1 ; sh:nodeKind sh:IRI ] ] + ) ] + [ sh:and ( + [ sh:property [ sh:path a:usageAttributionState ; sh:in ("no proven focus" "overlapping focus") ] ] + [ sh:property [ sh:path a:attributedTo ; sh:maxCount 0 ] ] + ) ] + ) . diff --git a/tests/test_usage_generation_budget.py b/tests/test_usage_generation_budget.py new file mode 100644 index 0000000..de20fb0 --- /dev/null +++ b/tests/test_usage_generation_budget.py @@ -0,0 +1,57 @@ +import sys +from pathlib import Path +import unittest +sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts')) +import usage_generation_budget as b + + +class BudgetTests(unittest.TestCase): + def test_part_boundaries_and_no_input_mutation(self): + old=[];got=b.admit('part',65536,40,3,100,old,staged_bytes=64*1024*1024) + self.assertEqual(old,[]);self.assertEqual(len(got),1) + for args in [('part',65537,40,3),('part',65536,41,3),('part',1,1,4), + ('marker',2049,0,2),('marker',2048,0,3),('graph',1,0,3)]: + with self.subTest(args=args),self.assertRaises(ValueError): b.admit(*args,100,[]) + with self.assertRaises(ValueError): b.admit('part',1,1,3,100,[],staged_bytes=64*1024*1024+1) + + def test_C_and_A_share_one_minute(self): + ledger=b.admit('whole',100,1,3,100,[]) + with self.assertRaises(ValueError): b.admit('part',100,1,3,159,ledger) + self.assertEqual(len(b.admit('part',100,1,3,160,ledger)),2) + with self.assertRaises(ValueError): b.admit('marker',100,0,2,99,ledger) + + def test_sixty_writes_and_hour_bytes(self): + ledger=[] + for n in range(60): ledger=b.admit('part',65536,40,3,n*60,ledger) + # At the hour boundary, the oldest entry expires; there is no burst allowance. + self.assertEqual(len(b.admit('part',65536,40,3,3600,ledger)),60) + with self.assertRaises(ValueError): b.admit('part',1,1,3,3599,ledger) + full=[{'at':100,'bytes':b.TOTAL_BYTES_HOUR,'kind':'whole'}] + with self.assertRaisesRegex(ValueError,'C\+A'): b.admit('marker',1,0,2,160,full) + full=[{'at':100,'bytes':b.A_BYTES_HOUR,'kind':'part'}] + with self.assertRaisesRegex(ValueError,'A body'): b.admit('marker',1,0,2,160,full) + + def test_unknown_ledgers_never_reset(self): + for ledger in [None,{},[{}],[{'at':float('nan'),'bytes':1,'kind':'part'}], + [{'at':200,'bytes':1,'kind':'part'}], + [{'at':20,'bytes':1,'kind':'part'},{'at':10,'bytes':1,'kind':'part'}]]: + with self.subTest(ledger=ledger),self.assertRaises(ValueError): b.admit('part',1,1,3,100,ledger) + + def test_session_scope_and_staging(self): + kw={'c_items':9,'source_digest':'a'*64,'previous':{},'staged_bytes':0,'body_bytes':65536} + self.assertTrue(b.admit_session(**kw)) + for change in [{'c_items':8},{'source_digest':'wrong'}, + {'previous':{'selected':{'source_digest':'a'*64}}}, + {'previous':{'in_progress':{'status':'staging'}}}, + {'previous':{'foreign':{}}}, + {'staged_bytes':64*1024*1024}]: + with self.subTest(change=change),self.assertRaises(ValueError): b.admit_session(**(kw|change)) + self.assertTrue(b.admit_session(**(kw|{'previous':{'in_progress':{'status':'abandoned'}}}))) + + def test_one_graph_read_per_build_five_minutes(self): + kw={'now':100,'previous':None,'graph':'https://example.org/g'} + first=b.admit_read(**kw) + with self.assertRaises(ValueError): b.admit_read(**(kw|{'now':399,'previous':first})) + self.assertEqual(b.admit_read(**(kw|{'now':400,'previous':first}))['at'],400) + for change in [{'requests':2},{'graph':'ROOT'},{'response_bytes':8*1024*1024+1},{'previous':{}}]: + with self.subTest(change=change),self.assertRaises(ValueError): b.admit_read(**(kw|change)) diff --git a/tests/test_usage_generation_codec.py b/tests/test_usage_generation_codec.py new file mode 100644 index 0000000..47e3bbf --- /dev/null +++ b/tests/test_usage_generation_codec.py @@ -0,0 +1,93 @@ +import copy +import json +from pathlib import Path +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts')) +import usage_generation_codec as c +from publish_work_cost import snapshots + + +class CodecTests(unittest.TestCase): + def manifest(self): + return {'version': c.VERSION, 'producer': 'https://example.org/producer', + 'harness': 'codex', 'session': 'fixture', 'horizon': '2026-10-24T00:00:00Z', + 'source_sha256': 'a'*64, 'previous': None, + 'parts': [[0, 'b'*64, 2, 65536]], 'records': 2} + + def validate(self, m): + return c.validate_manifest(c.canonical(m), producer='https://example.org/producer', + harness='codex', session='fixture', at='2026-10-24T00:00:00Z') + + def test_positive_and_empty(self): + self.assertEqual(self.validate(self.manifest())['records'], 2) + m = self.manifest(); m.update(parts=[], records=0) + self.assertEqual(self.validate(m)['records'], 0) + + def test_canonical_refusals(self): + for v in [float('nan'), float('inf'), 1.2, 2**53, '\ud800', {1: 'nonstring'}]: + with self.subTest(value=repr(v)), self.assertRaises((ValueError, UnicodeError)): + c.canonical(v) + self.assertEqual(c.canonical({'b': 1, 'a': 'é'}), '{"a":"é","b":1}') + + def test_strict_schema_counts_and_identity(self): + mutations = [('records', -1), ('records', True), ('records', 3), + ('source_sha256', 'BAD'), ('previous', 'relative'), + ('horizon', '2026-02-30T00:00:00Z'), ('session', 'other'), + ('harness', 'claude'), ('producer', 'https://example.org/other'), + ('version', 'future')] + for key, value in mutations: + m = self.manifest(); m[key] = value + with self.subTest(key=key), self.assertRaises(ValueError): + self.validate(m) + for parts in [[[0,'b'*64,41,1]], [[0,'b'*64,2,65537]], + [[1,'b'*64,2,1]], [[0,'b'*64,True,1]], + [[0,'b'*64,1,1],[0,'c'*64,1,1]]]: + m = self.manifest(); m['parts'] = parts + with self.subTest(parts=parts), self.assertRaises(ValueError): + self.validate(m) + m = self.manifest(); m['extra'] = 1 + with self.assertRaises(ValueError): self.validate(m) + + def test_noncanonical_and_duplicate_keys(self): + for raw in [json.dumps(self.manifest()), '{"version":1,"version":2}']: + with self.assertRaises(ValueError): + c.validate_manifest(raw, producer='https://example.org/producer', harness='codex', + session='fixture', at='2026-10-24T00:00:00Z') + + @staticmethod + def record(reason='no proven focus'): + return {'harness':'codex','session':'s','id':'r','tokens':12, + 'counts':{'input_uncached':10,'cache_read_input':0,'cache_write_input':0,'output':2}, + 'model':'fixture','model_source':'declared','at':'2026-10-24T00:00:00Z', + 'attribution':reason,'bead':'unattributed','agent':'fixture'} + + def test_stable_request_identity_independent_of_attribution(self): + a = self.record(); b = copy.deepcopy(a); b.update(attribution='attributed',bead='p-one') + self.assertEqual(c.record_iri(c.record_row(a)), c.record_iri(c.record_row(b))) + self.assertNotEqual(c.digest(c.record_row(a)), c.digest(c.record_row(b))) + b['harness'] = 'claude' + self.assertNotEqual(c.record_iri(c.record_row(a)), c.record_iri(c.record_row(b))) + + def test_paired_attribution_state_is_not_recoverable_from_current_rdf(self): + """Pin the schema gap: distinct contracted row hashes, identical native input RDF.""" + a = self.record('no proven focus'); b = self.record('overlapping focus') + with patch('publish_work_cost.planes.plane_for', return_value='https://example.org/observed'): + ta = next(snapshots({'errors':[], 'records':[a]}, 'fixture'))[0]['turtle'] + tb = next(snapshots({'errors':[], 'records':[b]}, 'fixture'))[0]['turtle'] + self.assertEqual(ta, tb) + self.assertNotEqual(c.digest([c.record_row(a)]), c.digest([c.record_row(b)])) + + def test_pinned_proposal_bytes(self): + root = Path(__file__).resolve().parents[1] + self.assertEqual(__import__('hashlib').sha256((root/'shapes/usage-generations.shapes.ttl').read_bytes()).hexdigest(), + 'fb04e6bd343022a8ccd0aa9d5006cbf2f126a8f4f160a1f201a5b63512573906') + + def test_prepare_rows_refuses_duplicate_or_mixed_identity(self): + a=self.record();b=copy.deepcopy(a);b['id']='r2' + self.assertEqual(len(c.prepare_rows([b,a])),2) + for records in [[a,a],[a,b|{'harness':'claude'}],[a,b|{'session':'other'}], + [a,b|{'agent':'other'}],[a|{'model':12}]]: + with self.subTest(records=records),self.assertRaises(ValueError):c.prepare_rows(records) diff --git a/tests/test_usage_generation_protocol.py b/tests/test_usage_generation_protocol.py new file mode 100644 index 0000000..e451eda --- /dev/null +++ b/tests/test_usage_generation_protocol.py @@ -0,0 +1,206 @@ +import copy +from datetime import datetime,timezone +import json +from pathlib import Path +import re +import sys +import tempfile +from unittest.mock import patch +import unittest +sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts')) +from test_usage_generation_codec import CodecTests +import usage_generation_codec as c +import usage_generation_wire as w +from usage_generation_reader import read_snapshot +import publish_work_cost as publisher + + +PRODUCER='https://example.org/producer' +AT='2026-10-24T00:00:00Z' +NOW=datetime(2026,10,24,tzinfo=timezone.utc).timestamp() + + +def records(n=12): + return [CodecTests.record('attributed')|{'id':str(i),'bead':'p-'+str(i)} for i in range(n)] + + +def planned(rows=None): + return w.plan(records() if rows is None else rows,producer=PRODUCER,harness='codex', + session='s',at=AT,actor='fixture') + + +def response(quads): + rows=[] + for g,s,p,o in sorted(set(quads),key=repr): + row={'s':s,'p':p,'lex':o[1],'iri':o[0]=='uri'} + if g is not None:row['g']=g + if o[0]=='literal':row.update(datatype=o[2],lang='') + rows.append(row) + return {'rows':rows,'count':len(rows),'truncated':False} + + +def all_response(p): + q=[(None,s,pred,tuple(o)) for s,pred,o in p['activation_triples']] + q += [(p['generation'],s,pred,tuple(o)) for block in [*p['part_triples'],p['metadata_triples']] for s,pred,o in block] + return response(q) + + +class ProtocolTests(unittest.TestCase): + def read(self,r):return read_snapshot(r,producer=PRODUCER,harness='codex',session='s',at=AT) + + def test_complete_oracle_and_all_attribution_states(self): + rows=records()+[CodecTests.record('no proven focus')|{'id':'unknown'}, + CodecTests.record('overlapping focus')|{'id':'overlap'}] + p=planned(rows);r=self.read(all_response(p)) + self.assertEqual(r['rows'],c.prepare_rows(rows));self.assertEqual(r['status'],'COMPLETE') + self.assertEqual(p['manifest']['source_sha256'],c.digest(r['rows'])) + + def test_missing_tampered_duplicate_wrong_kind_and_legacy_refuse(self): + original=all_response(planned()) + missing=copy.deepcopy(original);missing['rows'].pop();missing['count']-=1 + duplicate=copy.deepcopy(original);duplicate['rows'].append(duplicate['rows'][0]);duplicate['count']+=1 + changed=copy.deepcopy(original) + next(r for r in changed['rows'] if r['p']==c.ONTOLOGY+'tokensConsumed')['lex']='999' + kind=copy.deepcopy(original);next(r for r in kind['rows'] if r['p']==c.ONTOLOGY+'tokensConsumed')['datatype']=w.XSD+'string' + legacy=copy.deepcopy(original);next(r for r in legacy['rows'] if 'g' in r)['g']='https://example.org/legacy' + for r in [missing,duplicate,changed,kind,legacy,original|{'truncated':True}]: + with self.subTest(response=r),self.assertRaises(ValueError):self.read(r) + + def test_body_and_record_caps_with_single_oversize_refusal(self): + p=planned(records(81));self.assertEqual(sum(d[2] for d in p['manifest']['parts']),81) + self.assertTrue(all(d[2]<=40 and d[3]<=65536 for d in p['manifest']['parts'])) + self.assertTrue(all(w.byte_size(b)<=65536 for b in p['parts'])) + self.assertLessEqual(w.byte_size(p['activation']),2048) + with self.assertRaises(ValueError):planned([CodecTests.record()|{'model':'x'*65536}]) + with self.assertRaises(ValueError):planned([CodecTests.record()|{'at':'2026-10-25T00:00:00Z'}]) + empty=planned([]);self.assertEqual(self.read(all_response(empty))['rows'],[]) + + def test_reassignment_changes_generation_not_request_identity(self): + old=records();new=copy.deepcopy(old);new[0]['bead']='p-other' + p1=planned(old);p2=planned(new) + self.assertNotEqual(p1['generation'],p2['generation']) + self.assertEqual(c.record_iri(p1['rows'][0]),c.record_iri(p2['rows'][0])) + got=self.read(all_response(p2));self.assertNotIn(c.ONTOLOGY+'p-0',got['rows'][0][8]) + + +class FixtureTransport: + def __init__(self):self.graphs={};self.snapshots={};self.calls=[];self.labels={} + def __call__(self,ep,body,**kwargs): + from rdflib import Graph,URIRef,Literal + self.calls.append((ep,body)) + if ep=='/shapes':return {'turtle':(Path(__file__).resolve().parents[1]/'shapes/usage-generations.shapes.ttl').read_text()} + if ep=='/graph/create': + self.graphs.setdefault(body['graph'],set());return {'g':list(self.graphs).index(body['graph'])+1,'graph':body['graph']} + if ep=='/graph/label': + self.labels[body['graph']]={'kind':{'coverage':'full','value':['knowledge']}};return {'tx_id':len(self.calls)} + if ep=='/knot': + g=Graph().parse(data=body['turtle'],format='turtle');triples=set() + for s,p,o in g: + term=('uri',str(o),None) if isinstance(o,URIRef) else ('literal',str(o),str(o.datatype or URIRef(w.XSD+'string'))) + triples.add((str(s),str(p),term)) + self.snapshots[(body['graph'],body['snapshot'])]=triples + self.graphs[body['graph']]=set().union(*(t for (g,_),t in self.snapshots.items() if g==body['graph'])) + return {'conforms':True,'tx_id':len(self.calls)} + assert ep=='/query' + q=body['query'] + if 'VALUES ?item' in q: + items=re.findall(r'<([^>]+)>',q.split('VALUES ?item {')[1].split('}')[0]) + return {'rows':[{'item':i} for i in items],'truncated':False} + graph=body.get('graph') or re.search(r'GRAPH <([^>]+)>',q).group(1) + if 'usageSelectedGeneration' in q: + controls=self.graphs[graph];gens={o[1] for _,p,o in controls if p==c.ONTOLOGY+'usageSelectedGeneration'} + return response([(None,s,p,o) for s,p,o in controls]+[(g,s,p,o) for g in gens for s,p,o in self.graphs[g]]) + triples=self.graphs[graph] + if 'VALUES ?s' in q: + subjects=set(re.findall(r'<([^>]+)>',q.split('VALUES ?s {')[1].split('}')[0])) + return response([(None,s,p,o) for s,p,o in triples if s in subjects]) + return {'rows':[{'s':next(iter(triples))[0]}] if triples else [],'truncated':False,'labels':self.labels.get(graph,{})} + + +class PublisherTests(unittest.TestCase): + def setUp(self): + self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) + self.state=Path(self.tmp.name)/'state.json';self.transport=FixtureTransport() + auth=Path(self.tmp.name)/'authority.json';auth.write_text(json.dumps({'publish_work_cost.py':['read:crew:records']})) + self.env=patch.dict('os.environ',{'CAMAYOC_AUTHORITY':str(auth)});self.env.start();self.addCleanup(self.env.stop) + self.sleep=patch('time.sleep');self.sleep.start();self.addCleanup(self.sleep.stop) + self.policy={'enabled':True,'producer':PRODUCER,'harness':'codex','session':'s', + 'legacy_frozen':True,'runtime_acceptance':'fixture-only-not-an-enrollment'} + self.now=NOW + + def tick(self,rows): + self.now+=61 + before=len(self.transport.calls) + with patch('time.time',return_value=self.now),patch('planes._post',side_effect=self.transport): + answer=publisher.publish({'errors':[],'records':rows},'fixture',self.state,generation_policy=self.policy) + calls=self.transport.calls[before:] + self.assertLessEqual(len([e for e,_ in calls if e!='/query' and e!='/shapes']),1) + self.assertLessEqual(len(calls),3) + return answer + + def test_full_ticks_then_correction_then_retention_fence(self): + rows=records() + for _ in range(8):self.tick(rows) + journal=json.loads(self.state.with_suffix('.generations.json').read_text()) + entry=next(iter(journal['sessions'].values()));self.assertIsNone(entry['progress']);self.assertIsNotNone(entry['selected']) + old=entry['selected']['generation'];self.tick(rows) + with patch('time.time',return_value=self.now+61),patch('planes._post',side_effect=self.transport): + with self.assertRaisesRegex(ValueError,'frozen'): + publisher.publish({'errors':[],'records':rows[:2]},'fixture',self.state) + self.now+=61 + changed=copy.deepcopy(rows);changed[0]['bead']='p-other' + for _ in range(7):self.tick(changed) + journal=json.loads(self.state.with_suffix('.generations.json').read_text());entry=next(iter(journal['sessions'].values())) + self.assertNotEqual(entry['selected']['generation'],old);self.assertEqual(entry['retained'][0]['generation'],old) + changed[1]['bead']='p-third' + with self.assertRaisesRegex(ValueError,'retention'):self.tick(changed) + + def test_lost_local_cursor_uses_exact_verified_receipt_without_resend(self): + rows=records() + for _ in range(5):self.tick(rows) + journal=self.state.with_suffix('.generations.json');before=journal.read_bytes() + self.tick(rows) + journal.write_bytes(before) # private fixture simulates crash before cursor commit + calls=len(self.transport.calls) + self.assertEqual(self.tick(rows),[]) + self.assertEqual(len(self.transport.calls),calls) + receipt=json.loads(self.state.with_suffix('.receipt.json').read_text()) + self.assertTrue(receipt['recovered_cursor']);self.assertEqual(receipt['requests'],0) + self.assertEqual(receipt['request_seconds'],0);self.assertEqual(receipt['knot_seconds'],0) + + def test_policy_and_pending_fences(self): + self.policy['runtime_acceptance']='' + with self.assertRaises(ValueError):self.tick(records()) + self.assertEqual(self.transport.calls,[]) + self.policy['runtime_acceptance']='fixture' + self.state.with_suffix('.pending.json').write_text('{"preserve":"unknown"}') + with self.assertRaisesRegex(ValueError,'INDETERMINATE'):self.tick(records()) + self.assertEqual(self.transport.calls,[]) + + +class ConsumerTests(unittest.TestCase): + def test_global_consumer_attempt_and_failure_preserve_unknown(self): + from usage_generation_reader import consume + with tempfile.TemporaryDirectory() as d: + state=Path(d)/'consumer.json';calls=[] + def post(*args,**kwargs):calls.append(args);return all_response(planned()) + args={'producer':PRODUCER,'harness':'codex','session':'s','at':AT,'state_path':state,'post':post} + first=consume(**args,now=NOW);self.assertEqual(first['status'],'COMPLETE') + second=consume(**(args|{'session':'other'}),now=NOW+1) + self.assertEqual(second['status'],'UNKNOWN');self.assertEqual(len(calls),1) + third=consume(**args,now=NOW+300);self.assertEqual(third['status'],'COMPLETE') + self.assertEqual(len(calls),2) + + +class RetrievalRouteTests(unittest.TestCase): + def test_generation_method_scope_and_fixed_consumer_cache(self): + from retrieve_metric import execute + params={'producer':PRODUCER,'harness':'codex','session':'s','horizon':AT,'consumer':'brief'} + with tempfile.TemporaryDirectory() as d,patch.dict('os.environ',{'CAMAYOC_COST_READ_STATE':d}),patch('usage_generation_reader.consume',return_value={'status':'UNKNOWN'}) as consume: + result=execute({'system':'session_usage','query':'work_cost_generation','params':params}) + self.assertEqual(result['status'],'unknown') + self.assertEqual(consume.call_args.kwargs['state_path'],Path(d)/'brief.json') + consume.reset_mock() + for bad in [params|{'consumer':'new-poller'},params|{'state_path':'arbitrary'}]: + self.assertEqual(execute({'system':'session_usage','query':'work_cost_generation','params':bad})['status'],'invalid') + consume.assert_not_called() diff --git a/tests/test_usage_generation_step.py b/tests/test_usage_generation_step.py new file mode 100644 index 0000000..2758ec0 --- /dev/null +++ b/tests/test_usage_generation_step.py @@ -0,0 +1,79 @@ +import json +import sys +import tempfile +from pathlib import Path +from unittest.mock import patch +import unittest +sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts')) +import usage_generation_step as s + + +class StepTests(unittest.TestCase): + def setUp(self): + self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) + self.state=Path(self.tmp.name)/'state.json';self.calls=[] + self.sleep=patch.object(s.time,'sleep');self.sleep.start();self.addCleanup(self.sleep.stop) + + def post(self, endpoint, body): + self.calls.append((endpoint,body)) + if endpoint!='/query': + pending=self.state.with_suffix('.pending.json') + self.assertTrue(pending.exists()) + self.assertEqual(json.loads(pending.read_text())['body'],body) + return {'tx_id':0,'conforms':True} + return {'rows':[{'id':'positive'}]} + + def test_three_requests_zero_tx_and_rate_shared(self): + r=s.run_step('part',{'turtle':'fixture'},self.state,now=100,post=self.post, + records=1,preflight=lambda call:bool(call('/query',{})['rows']), + verify=lambda call,r:bool(call('/query',{})['rows'])) + self.assertEqual(r['status'],'VERIFIED');self.assertEqual(len(self.calls),3) + self.assertFalse(self.state.with_suffix('.pending.json').exists()) + with self.assertRaises(ValueError): + s.run_step('whole',{},self.state,now=159,post=self.post,verify=lambda *_:True) + self.assertEqual(len(self.calls),3) + + def test_marker_two_requests(self): + r=s.run_step('marker',{},self.state,now=100,post=self.post, + verify=lambda call,r:bool(call('/query',{})['rows'])) + self.assertEqual(r['requests'],2) + + def test_lost_reply_never_resends_or_drops_pending(self): + def lost(endpoint,body): self.post(endpoint,body);raise TimeoutError('lost reply') + with self.assertRaises(TimeoutError): + s.run_step('part',{},self.state,now=100,post=lost,records=1,verify=lambda *_:True) + before=self.state.with_suffix('.pending.json').read_bytes() + with self.assertRaisesRegex(ValueError,'INDETERMINATE'): + s.run_step('part',{},self.state,now=1000,post=self.post,records=1,verify=lambda *_:True) + self.assertEqual(len(self.calls),1) + self.assertEqual(before,self.state.with_suffix('.pending.json').read_bytes()) + self.assertEqual(len(json.loads(self.state.with_suffix('.writes.json').read_text())),1) + + def test_refusal_and_readback_mismatch_stall(self): + for accepted in [{'conforms':False},{},{'tx_id':1}]: + with tempfile.TemporaryDirectory() as d: + state=Path(d)/'state.json' + with self.assertRaises(ValueError): + s.run_step('part',{},state,now=100,post=lambda *_:accepted, + records=1,verify=lambda *_:False) + self.assertTrue(state.with_suffix('.pending.json').exists()) + + def test_budget_and_preflight_refuse_before_write(self): + for size in [65537,70000]: + with self.assertRaises(ValueError): + s.run_step('part',{'turtle':'x'*size},self.state,now=100,post=self.post, + records=1,verify=lambda *_:True) + with self.assertRaises(ValueError): + s.run_step('part',{},self.state,now=100,post=self.post,records=1, + preflight=lambda _:False,verify=lambda *_:True) + self.assertEqual(self.calls,[]) + self.assertFalse(self.state.with_suffix('.pending.json').exists()) + + def test_exclusive_writer_and_corrupt_ledger(self): + with s.locked(self.state): + with self.assertRaisesRegex(ValueError,'busy'): + s.run_step('part',{},self.state,now=100,post=self.post,records=1,verify=lambda *_:True) + self.state.with_suffix('.writes.json').write_text('{bad') + with self.assertRaises(ValueError): + s.run_step('part',{},self.state,now=100,post=self.post,records=1,verify=lambda *_:True) + self.assertEqual(self.calls,[]) diff --git a/tests/test_usage_generation_terms.py b/tests/test_usage_generation_terms.py new file mode 100644 index 0000000..29a7e16 --- /dev/null +++ b/tests/test_usage_generation_terms.py @@ -0,0 +1,40 @@ +import sys +from pathlib import Path +import unittest +sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts')) +import usage_generation_terms as t + + +class TermTests(unittest.TestCase): + def payload(self,**changes): + row={'s':'https://example.org/s','p':'https://example.org/p','lex':'https://example.org/value', + 'iri':False,'datatype':t.XSD+'string','lang':''} + return {'rows':[row|changes],'truncated':False,'count':1} + + def test_iri_looking_literal_is_not_iri(self): + literal=t.parse_response(self.payload(),'https://example.org/g')[0][-1] + raw=self.payload();raw['rows'][0]={'s':'https://example.org/s','p':'https://example.org/p', + 'lex':'https://example.org/value','iri':True} + real=t.parse_response(raw,'https://example.org/g')[0][-1] + self.assertNotEqual(literal,real) + + def test_kind_language_datatype_and_duplicate_refusals(self): + for changes in [{'iri':None},{'iri':True},{'lang':'en'}, + {'datatype':t.XSD+'double'},{'lex':12},{'g':'ROOT'}, + {'extra':'bad'}]: + with self.subTest(changes=changes),self.assertRaises(ValueError): + t.parse_response(self.payload(**changes),'https://example.org/g') + duplicate=self.payload();duplicate['rows']*=2;duplicate['count']=2 + with self.assertRaises(ValueError):t.parse_response(duplicate,'https://example.org/g') + + def test_unknown_empty_and_truncated_are_not_complete(self): + for response in [{}, {'rows':[],'truncated':False,'count':0}, + self.payload()|{'truncated':True},self.payload()|{'count':False}]: + with self.subTest(response=response),self.assertRaises(ValueError): + t.parse_response(response,'https://example.org/g') + + def test_one_query_names_scope_and_has_no_root_fallback(self): + q=t.snapshot_query('https://example.org/control','https://example.org/selector') + self.assertTrue(q['verbose']);self.assertIn('GRAPH ',q['query']) + self.assertIn('GRAPH ?g',q['query']);self.assertIn('STR(?o)',q['query']) + with self.assertRaises(ValueError):t.snapshot_query('ROOT','relative') From 8f63e9eddb26695315734997ba30d58dbd6e68cc Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Fri, 23 Oct 2026 21:11:20 -0400 Subject: [PATCH 2/2] fix(cost): validate staged journals and retained storage bounds --- scripts/usage_generation_publisher.py | 53 +++++++++++++++++++++++-- tests/test_usage_generation_protocol.py | 10 +++++ 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/scripts/usage_generation_publisher.py b/scripts/usage_generation_publisher.py index 2c54693..8cfff97 100644 --- a/scripts/usage_generation_publisher.py +++ b/scripts/usage_generation_publisher.py @@ -30,11 +30,54 @@ def policy_check(policy): def load_state(path): if not path.exists():return {'version':1,'sessions':{},'profile':None} value=json.loads(path.read_text()) - if not isinstance(value,dict) or set(value)!={'version','sessions','profile'} or value['version']!=1 or not isinstance(value['sessions'],dict): + if not isinstance(value,dict) or set(value)!={'version','sessions','profile'} or type(value['version']) is not int or value['version']!=1 or not isinstance(value['sessions'],dict): raise ValueError('UNKNOWN: invalid generation journal') + if value['profile'] is not None and (not isinstance(value['profile'],str) or not c.HEX.fullmatch(value['profile'])): + raise ValueError('UNKNOWN: invalid profile proof') + for key,entry in value['sessions'].items(): + if not isinstance(entry,dict) or set(entry)!={'selected','progress','retained'} or not isinstance(entry['retained'],list) or len(entry['retained'])>1: + raise ValueError('UNKNOWN: invalid generation slots') + for selected in [entry['selected'],*entry['retained']]: + if selected is None:continue + required={'generation','source','horizon','bytes','producer','harness','session'} + if not isinstance(selected,dict) or set(selected)!=required: + raise ValueError('UNKNOWN: invalid sealed receipt') + c.iri(selected['generation']);c.iri(selected['producer']);c.text(selected['session']);c.horizon(selected['horizon']) + if (selected['harness'] not in ('claude','codex') or not isinstance(selected['source'],str) + or not c.HEX.fullmatch(selected['source']) or c.count(selected['bytes'])>c.STAGED_BYTES + or key!=c.digest([selected['producer'],selected['harness'],selected['session']])): + raise ValueError('UNKNOWN: sealed identity/bytes mismatch') + progress=entry['progress'] + if progress is not None: + if not isinstance(progress,dict) or set(progress)!={'plan','plan_sha256','phase','part','receipts'}: + raise ValueError('UNKNOWN: invalid staging receipt') + actual=hashlib.sha256(json.dumps(progress['plan'],sort_keys=True).encode()).hexdigest() + if actual!=progress['plan_sha256'] or not isinstance(progress['receipts'],list): + raise ValueError('UNKNOWN: staging artifact hash differs') + planned=progress['plan'] + if not isinstance(planned,dict) or not isinstance(planned.get('manifest'),dict): + raise ValueError('UNKNOWN: invalid generation artifact') + m=planned['manifest'] + c.validate_manifest(c.canonical(m),producer=m.get('producer'),harness=m.get('harness'),session=m.get('session'),at=m.get('horizon')) + if (key!=c.digest([m['producer'],m['harness'],m['session']]) + or planned.get('generation')!=c.BASE+'generation/'+c.digest(m) + or planned.get('control')!=c.BASE+'selector-graph/'+key + or planned.get('selector')!=c.BASE+'selector/'+key + or c.count(planned.get('bytes'))>c.STAGED_BYTES + or len(planned.get('parts',[]))!=len(m['parts']) + or progress['phase'] not in ('control','control-label','graph','graph-label','part','manifest','activate') + or c.count(progress['part'])>len(m['parts']) + or progress['phase']=='part' and progress['part']==len(m['parts'])): + raise ValueError('UNKNOWN: staging identity/phase mismatch') return value +def unselected_bytes(state): + return sum((entry['progress']['plan']['bytes'] if entry.get('progress') else 0) + +sum(retained['bytes'] for retained in entry['retained']) + for entry in state['sessions'].values()) + + def query_triples(graph,subjects): values=' '.join('<'+c.iri(s)+'>' for s in sorted(subjects)) return {'query':'SELECT ?s ?p (STR(?o) AS ?lex) (isIRI(?o) AS ?iri) ' @@ -97,9 +140,9 @@ def post(endpoint,body): planned=plan(result['records'],producer=policy['producer'],harness=policy['harness'], session=policy['session'],at=horizon,actor=actor, previous=selected['generation'] if selected else None) - total=sum(e['progress']['plan']['bytes'] for e in state['sessions'].values() if e.get('progress')) + total=unselected_bytes(state) if total+planned['bytes']>c.STAGED_BYTES:raise ValueError('aggregate staged bytes exceed64MiB') - progress={'plan':planned,'phase':'control','part':0,'receipts':[]} + progress={'plan':planned,'plan_sha256':hashlib.sha256(json.dumps(planned,sort_keys=True).encode()).hexdigest(),'phase':'control','part':0,'receipts':[]} entry['progress']=progress;state['sessions'][key]=entry;atomic(journal_path,state) planned=progress['plan'];phase=progress['phase'];index=progress['part'] if phase in ('control','graph'): @@ -133,6 +176,8 @@ def preflight(call): body=planned['metadata'];kind='part';n=0;preflight=None verify=lambda call,_:verify_triples(call,planned['generation'],planned['metadata_triples']) elif phase=='activate': + after=unselected_bytes(state)-planned['bytes']+(entry['selected']['bytes'] if entry['selected'] else 0) + if after>c.STAGED_BYTES:raise ValueError('unselected retention would exceed64MiB') body=planned['activation'];kind='marker';n=0;preflight=None def verify(call,_): checked=read_snapshot(call('/query',snapshot_query(planned['control'],planned['selector'])), @@ -152,7 +197,7 @@ def verify(call,_): accepted=prior else: accepted=run_step(kind,body,state_path,now=now,post=post,verify=verify, - preflight=preflight,records=n,lane_locked=lane_locked,staged_bytes=sum(e['progress']['plan']['bytes'] for e in state['sessions'].values() if e.get('progress'))) + preflight=preflight,records=n,lane_locked=lane_locked,staged_bytes=unselected_bytes(state)) progress['receipts'].append({'phase':phase,'index':index,'sha256':body_hash, 'bytes':byte_size(body),'accepted':accepted.get('accepted')}) if phase=='control':progress['phase']='control-label' diff --git a/tests/test_usage_generation_protocol.py b/tests/test_usage_generation_protocol.py index e451eda..833691c 100644 --- a/tests/test_usage_generation_protocol.py +++ b/tests/test_usage_generation_protocol.py @@ -168,6 +168,16 @@ def test_lost_local_cursor_uses_exact_verified_receipt_without_resend(self): self.assertTrue(receipt['recovered_cursor']);self.assertEqual(receipt['requests'],0) self.assertEqual(receipt['request_seconds'],0);self.assertEqual(receipt['knot_seconds'],0) + def test_corrupt_artifact_and_forged_selected_state_refuse_without_requests(self): + rows=records();self.tick(rows);self.tick(rows) + path=self.state.with_suffix('.generations.json');original=path.read_text() + j=json.loads(original);entry=next(iter(j['sessions'].values())) + entry['progress']['plan']['manifest']['records']+=1 + path.write_text(json.dumps(j));before=len(self.transport.calls) + with self.assertRaises(ValueError):self.tick(rows) + self.assertEqual(before,len(self.transport.calls)) + path.write_text(original) + def test_policy_and_pending_fences(self): self.policy['runtime_acceptance']='' with self.assertRaises(ValueError):self.tick(records())