From 1b2615d27cdb23f82cd94324cba07a4ac9bb69d8 Mon Sep 17 00:00:00 2001 From: golaraj Date: Fri, 14 Aug 2026 10:11:19 -0700 Subject: [PATCH 1/2] Forge: persist sanitized fingerprint signatures Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: b8cf0691-0fe3-4d67-92ce-ce2a529c400c --- .../skills/repository-skill-forge/SKILL.md | 7 +- .../assets/schemas.json | 16 ++ .../scripts/aggregate-primitives.py | 80 +++++++- .../scripts/issue-state.py | 106 ++++++++++- .../tests/test_state_catalog.py | 172 ++++++++++++++++++ 5 files changed, 376 insertions(+), 5 deletions(-) create mode 100644 plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/SKILL.md b/plugins/repo-dreamer/skills/repository-skill-forge/SKILL.md index a5282d2..ad7b604 100644 --- a/plugins/repo-dreamer/skills/repository-skill-forge/SKILL.md +++ b/plugins/repo-dreamer/skills/repository-skill-forge/SKILL.md @@ -71,6 +71,7 @@ repository-skill-forge-state:v2:begin "cursor": null, "updatedAt": "", "observations": [], + "fingerprintCatalog": {}, "proposalQueue": [], "proposalHistory": {} } @@ -210,7 +211,11 @@ evidence into candidate generation. Raw commands and source content remain ephemeral. Persistent observations keep only evidence/fingerprint hashes, repository-salted session and branch hashes, timestamps, outcome, surface, kind, bounded path families, and repository -references. +references. Durable state also keeps a bounded catalog of versioned, sanitized +structural signatures keyed by fingerprint. This allows later runs to explain +and aggregate historical patterns without retaining raw commands or source +content. Legacy v2 states without `fingerprintCatalog` migrate as an empty +catalog on their next parse and render. ### 6. Build compact state and candidates diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/assets/schemas.json b/plugins/repo-dreamer/skills/repository-skill-forge/assets/schemas.json index edfec60..8fa94d8 100644 --- a/plugins/repo-dreamer/skills/repository-skill-forge/assets/schemas.json +++ b/plugins/repo-dreamer/skills/repository-skill-forge/assets/schemas.json @@ -259,6 +259,22 @@ "type": "array", "items": { "$ref": "#/$defs/compactObservation" } }, + "fingerprintCatalog": { + "type": "object", + "maxProperties": 64, + "propertyNames": { "pattern": "^[0-9a-f]{16}$" }, + "additionalProperties": { + "type": "object", + "required": ["kind", "signatureVersion", "signature", "lastSeenAt"], + "properties": { + "kind": { "enum": ["command", "script"] }, + "signatureVersion": { "const": 1 }, + "signature": { "type": "object", "minProperties": 1 }, + "lastSeenAt": { "type": "string" } + }, + "additionalProperties": false + } + }, "proposalQueue": { "type": "array", "items": { "$ref": "#/$defs/proposalQueueEntry" } diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py index 7b4bb9a..0cf4a3b 100755 --- a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py +++ b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py @@ -6,6 +6,7 @@ from __future__ import annotations import argparse +import json import re from collections import defaultdict from datetime import timedelta @@ -13,6 +14,11 @@ from forge_common import parse_timestamp, read_json, stable_hash, timestamp_text, write_json +SIGNATURE_VERSION = 1 +MAX_FINGERPRINT_CATALOG_ENTRIES = 64 +MAX_FINGERPRINT_CATALOG_BYTES = 12_000 +MAX_SIGNATURE_BYTES = 2_000 + def reference_values(primitive: dict[str, Any], ref_type: str) -> set[str]: refs = primitive.get("refs") @@ -34,10 +40,26 @@ def merge_evidence( if isinstance(previous_scope, dict) and previous_scope.get("repository") != repository: raise ValueError("state repository does not match the current repository") evidence_by_key: dict[str, dict[str, Any]] = {} + fingerprint_catalog = ( + state.get("fingerprintCatalog", {}) + if isinstance(state, dict) + else {} + ) + fingerprint_catalog = ( + fingerprint_catalog if isinstance(fingerprint_catalog, dict) else {} + ) if isinstance(state, dict): for item in state.get("observations", state.get("evidenceLedger", [])): if isinstance(item, dict) and item.get("evidenceKey"): - evidence_by_key[str(item["evidenceKey"])] = item + restored = dict(item) + catalog_entry = fingerprint_catalog.get(str(item.get("fingerprint"))) + if ( + isinstance(catalog_entry, dict) + and catalog_entry.get("kind") == item.get("kind") + and isinstance(catalog_entry.get("signature"), dict) + ): + restored["signature"] = catalog_entry["signature"] + evidence_by_key[str(item["evidenceKey"])] = restored for item in document.get("primitives", []): if isinstance(item, dict) and item.get("evidenceKey"): evidence_by_key[str(item["evidenceKey"])] = compact_observation( @@ -51,6 +73,61 @@ def merge_evidence( return list(evidence_by_key.values()), history if isinstance(history, dict) else {} +def build_fingerprint_catalog( + evidence: list[dict[str, Any]], +) -> dict[str, dict[str, Any]]: + entries: dict[str, dict[str, Any]] = {} + for item in evidence: + fingerprint = item.get("fingerprint") + kind = item.get("kind") + signature = item.get("signature") + completed_at = item.get("completedAt") + if ( + not isinstance(fingerprint, str) + or kind not in {"command", "script"} + or not isinstance(signature, dict) + or not signature + or not isinstance(completed_at, str) + or stable_hash({"kind": kind, "signature": signature}) != fingerprint + ): + continue + signature_bytes = len( + json.dumps(signature, sort_keys=True, separators=(",", ":")).encode() + ) + if signature_bytes > MAX_SIGNATURE_BYTES: + continue + current = entries.get(fingerprint) + if current is None or completed_at > current["lastSeenAt"]: + entries[fingerprint] = { + "kind": kind, + "signatureVersion": SIGNATURE_VERSION, + "signature": signature, + "lastSeenAt": completed_at, + } + + retained: dict[str, dict[str, Any]] = {} + retained_bytes = 2 + ordered = sorted( + entries.items(), + key=lambda item: (-parse_timestamp(item[1]["lastSeenAt"]).timestamp(), item[0]), + ) + for fingerprint, entry in ordered: + if len(retained) >= MAX_FINGERPRINT_CATALOG_ENTRIES: + break + entry_bytes = len( + json.dumps( + {fingerprint: entry}, + sort_keys=True, + separators=(",", ":"), + ).encode() + ) + if retained_bytes + entry_bytes > MAX_FINGERPRINT_CATALOG_BYTES: + continue + retained[fingerprint] = entry + retained_bytes += entry_bytes + return dict(sorted(retained.items())) + + def compact_observation(item: dict[str, Any], repository: str) -> dict[str, Any]: session_hash = ( str(item["sessionHash"]) @@ -347,6 +424,7 @@ def main() -> None: retained_observations, key=lambda item: str(item.get("evidenceKey")), ), + "fingerprintCatalog": build_fingerprint_catalog(retained_evidence), "proposalQueue": proposal_queue if isinstance(proposal_queue, list) else [], "proposalHistory": proposal_history, } diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py index a27b6b6..43f0800 100644 --- a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py +++ b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py @@ -12,7 +12,7 @@ from pathlib import Path from typing import Any -from forge_common import parse_timestamp, read_json, write_json +from forge_common import parse_timestamp, read_json, stable_hash, write_json MARKER = "repository-skill-forge-state" BLOCK_RE = re.compile( @@ -45,9 +45,11 @@ "cursor", "updatedAt", "observations", + "fingerprintCatalog", "proposalQueue", "proposalHistory", } +LEGACY_TOP_LEVEL = ALLOWED_TOP_LEVEL - {"fingerprintCatalog"} OBSERVATION_KEYS = { "evidenceKey", "fingerprint", @@ -111,8 +113,16 @@ FORBIDDEN_VALUE_RE = re.compile( r"(?:/(?:Users|home)/[^/\s]+|[A-Za-z]:\\Users\\|" r"\b(?:gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,})\b|" - r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----)" + r"\bAKIA[0-9A-Z]{16}\b|" + r"\bBearer\s+[A-Za-z0-9._~+/=-]{20,}|" + r"\b(?:password|passwd|token|secret|api[_-]?key)\s*[:=]\s*" + r"['\"]?(?!<|\$\{|\$[A-Z_]+)[^\s'\"]{12,}|" + r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----)", + re.IGNORECASE, ) +MAX_FINGERPRINT_CATALOG_ENTRIES = 64 +MAX_FINGERPRINT_CATALOG_BYTES = 12_000 +MAX_SIGNATURE_BYTES = 2_000 def validate_timestamp(value: Any, field: str) -> None: @@ -196,6 +206,90 @@ def validate_observations(observations: list[Any]) -> None: raise ValueError("issue state observation contains an invalid ref") +def validate_fingerprint_catalog(catalog: Any) -> None: + if not isinstance(catalog, dict): + raise ValueError("issue state fingerprintCatalog must be an object") + if len(catalog) > MAX_FINGERPRINT_CATALOG_ENTRIES: + raise ValueError("issue state fingerprintCatalog exceeds its entry limit") + serialized_bytes = len( + json.dumps(catalog, sort_keys=True, separators=(",", ":")).encode() + ) + if serialized_bytes > MAX_FINGERPRINT_CATALOG_BYTES: + raise ValueError("issue state fingerprintCatalog exceeds its size limit") + for fingerprint, entry in catalog.items(): + if not isinstance(fingerprint, str) or not re.fullmatch( + r"[0-9a-f]{16}", fingerprint + ): + raise ValueError("issue state fingerprintCatalog contains an invalid key") + if not isinstance(entry, dict) or set(entry) != { + "kind", + "signatureVersion", + "signature", + "lastSeenAt", + }: + raise ValueError( + "issue state fingerprintCatalog contains an unsupported entry" + ) + kind = entry.get("kind") + signature = entry.get("signature") + if kind not in {"command", "script"}: + raise ValueError("issue state fingerprintCatalog contains an invalid kind") + if ( + isinstance(entry.get("signatureVersion"), bool) + or entry.get("signatureVersion") != 1 + ): + raise ValueError( + "issue state fingerprintCatalog contains an unsupported signature version" + ) + if not isinstance(signature, dict) or not signature: + raise ValueError( + "issue state fingerprintCatalog requires a nonempty signature" + ) + signature_bytes = len( + json.dumps(signature, sort_keys=True, separators=(",", ":")).encode() + ) + if signature_bytes > MAX_SIGNATURE_BYTES: + raise ValueError("issue state fingerprintCatalog signature is too large") + if kind == "command": + if set(signature) != {"tokens"}: + raise ValueError( + "issue state command fingerprint requires a token signature" + ) + tokens = signature.get("tokens") + if ( + not isinstance(tokens, list) + or not tokens + or len(tokens) > 40 + or any(not isinstance(token, str) or len(token) > 80 for token in tokens) + ): + raise ValueError( + "issue state command fingerprint has invalid tokens" + ) + else: + if set(signature) != {"imports", "calls", "fileExtensions"}: + raise ValueError( + "issue state script fingerprint has an invalid signature" + ) + for key in ("imports", "calls", "fileExtensions"): + values = signature.get(key) + if ( + not isinstance(values, list) + or len(values) > 64 + or any( + not isinstance(value, str) or not value or len(value) > 200 + for value in values + ) + ): + raise ValueError( + f"issue state script fingerprint has invalid {key}" + ) + if stable_hash({"kind": kind, "signature": signature}) != fingerprint: + raise ValueError( + "issue state fingerprintCatalog signature does not match its key" + ) + validate_timestamp(entry.get("lastSeenAt"), "fingerprint lastSeenAt") + + def validate_reconciliation(value: Any) -> None: if value is None: return @@ -335,8 +429,13 @@ def validate_sanitized_value(value: Any, key: str | None = None) -> None: def validate_state(state: Any, repository: str) -> dict[str, Any]: if not isinstance(state, dict): raise ValueError("issue state must be a JSON object") - if set(state) != ALLOWED_TOP_LEVEL: + if frozenset(state) not in { + frozenset(ALLOWED_TOP_LEVEL), + frozenset(LEGACY_TOP_LEVEL), + }: raise ValueError("issue state has an unsupported top-level shape") + state = dict(state) + state.setdefault("fingerprintCatalog", {}) if state.get("schemaVersion") != 2: raise ValueError("unsupported issue state schema version") if not isinstance(state.get("stateVersion"), int) or state["stateVersion"] < 1: @@ -354,6 +453,7 @@ def validate_state(state: Any, repository: str) -> dict[str, Any]: if not isinstance(state.get("proposalHistory"), dict): raise ValueError("issue state proposalHistory must be an object") validate_observations(state["observations"]) + validate_fingerprint_catalog(state["fingerprintCatalog"]) validate_proposal_queue(state["proposalQueue"]) validate_proposal_history(state["proposalHistory"]) validate_sanitized_value(state) diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py b/plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py new file mode 100644 index 0000000..3c1fcd7 --- /dev/null +++ b/plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import importlib.util +import json +import sys +import unittest +from pathlib import Path + +SKILL_DIR = Path(__file__).resolve().parents[1] +SCRIPTS_DIR = SKILL_DIR / "scripts" +sys.path.insert(0, str(SCRIPTS_DIR)) + +from forge_common import stable_hash + + +def load_script(name: str, filename: str): + spec = importlib.util.spec_from_file_location(name, SCRIPTS_DIR / filename) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +aggregate = load_script("aggregate_primitives", "aggregate-primitives.py") +issue_state = load_script("issue_state", "issue-state.py") + + +def empty_state() -> dict[str, object]: + return { + "schemaVersion": 2, + "stateVersion": 1, + "scope": {"kind": "repository", "repository": "owner/repository"}, + "cursor": None, + "updatedAt": "2026-08-14T00:00:00Z", + "observations": [], + "proposalQueue": [], + "proposalHistory": {}, + } + + +def observation(fingerprint: str) -> dict[str, object]: + return { + "evidenceKey": "a" * 24, + "fingerprint": fingerprint, + "sessionHash": "b" * 24, + "completedAt": "2026-08-13T00:00:00Z", + "day": "2026-08-13", + "outcome": "success", + "surface": "cli", + "kind": "command", + "branchHash": None, + "branchCategory": "unknown", + "pathFamilies": [], + "refs": [], + } + + +class FingerprintCatalogTests(unittest.TestCase): + def test_legacy_state_migrates_with_empty_catalog(self) -> None: + body = ( + "repository-skill-forge-state:v2:begin\n" + f"{json.dumps(empty_state())}\n" + "repository-skill-forge-state:v2:end" + ) + + parsed = issue_state.parse_body(body, "owner/repository", 60_000) + + self.assertEqual({}, parsed["fingerprintCatalog"]) + + def test_catalog_restores_historical_signature_for_aggregation(self) -> None: + signature = {"tokens": ["git", "show", ""]} + fingerprint = stable_hash({"kind": "command", "signature": signature}) + state = empty_state() | { + "observations": [observation(fingerprint)], + "fingerprintCatalog": { + fingerprint: { + "kind": "command", + "signatureVersion": 1, + "signature": signature, + "lastSeenAt": "2026-08-13T00:00:00Z", + } + }, + } + + evidence, _history = aggregate.merge_evidence( + state, + {"primitives": []}, + "owner/repository", + ) + patterns = aggregate.aggregate( + evidence, + as_of="2026-08-14T00:00:00Z", + active_days=90, + stale_days=180, + merged_prs=set(), + thresholds={ + "minDistinctSessions": 1, + "minDistinctDays": 1, + "minKnownOutcomes": 1, + "minSuccessRate": 0, + "minScoredCoverage": 0, + "minMergedPrs": 0, + "minMainlineEvidence": 0, + }, + ) + + self.assertEqual(signature, patterns[0]["signature"]) + + def test_catalog_rejects_secret_shaped_signature_values(self) -> None: + signature = {"tokens": ["deploy", "token=abcdefghijklmnop"]} + fingerprint = stable_hash({"kind": "command", "signature": signature}) + state = empty_state() | { + "fingerprintCatalog": { + fingerprint: { + "kind": "command", + "signatureVersion": 1, + "signature": signature, + "lastSeenAt": "2026-08-13T00:00:00Z", + } + } + } + + with self.assertRaisesRegex(ValueError, "secret-shaped"): + issue_state.validate_state(state, "owner/repository") + + def test_catalog_rejects_signature_key_mismatch(self) -> None: + signature = {"tokens": ["git", "status"]} + state = empty_state() | { + "fingerprintCatalog": { + "c" * 16: { + "kind": "command", + "signatureVersion": 1, + "signature": signature, + "lastSeenAt": "2026-08-13T00:00:00Z", + } + } + } + + with self.assertRaisesRegex(ValueError, "does not match"): + issue_state.validate_state(state, "owner/repository") + + def test_catalog_builder_enforces_entry_and_byte_limits(self) -> None: + evidence = [] + for index in range(100): + signature = {"tokens": ["tool", f"operation-{index}"]} + evidence.append( + { + "fingerprint": stable_hash( + {"kind": "command", "signature": signature} + ), + "kind": "command", + "signature": signature, + "completedAt": f"2026-08-13T{index % 24:02d}:00:00Z", + } + ) + + catalog = aggregate.build_fingerprint_catalog(evidence) + + self.assertLessEqual( + len(catalog), + aggregate.MAX_FINGERPRINT_CATALOG_ENTRIES, + ) + self.assertLessEqual( + len(json.dumps(catalog, separators=(",", ":")).encode()), + aggregate.MAX_FINGERPRINT_CATALOG_BYTES, + ) + + +if __name__ == "__main__": + unittest.main() From 2b2d074893ec286faf4ea42b409f6ab39b25b504 Mon Sep 17 00:00:00 2001 From: golaraj Date: Fri, 14 Aug 2026 14:34:15 -0700 Subject: [PATCH 2/2] Forge: validate fingerprint catalog boundaries Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: b8cf0691-0fe3-4d67-92ce-ce2a529c400c --- .../scripts/aggregate-primitives.py | 33 +++++++- .../scripts/issue-state.py | 5 +- .../tests/test_state_catalog.py | 82 +++++++++++++++++++ 3 files changed, 117 insertions(+), 3 deletions(-) diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py index 0cf4a3b..bc15bf3 100755 --- a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py +++ b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/aggregate-primitives.py @@ -6,10 +6,12 @@ from __future__ import annotations import argparse +import importlib.util import json import re from collections import defaultdict from datetime import timedelta +from pathlib import Path from typing import Any from forge_common import parse_timestamp, read_json, stable_hash, timestamp_text, write_json @@ -20,6 +22,22 @@ MAX_SIGNATURE_BYTES = 2_000 +def validate_next_state( + state: dict[str, Any], + repository: str, +) -> dict[str, Any]: + validator_path = Path(__file__).with_name("issue-state.py") + spec = importlib.util.spec_from_file_location( + "repository_skill_forge_issue_state", + validator_path, + ) + if spec is None or spec.loader is None: + raise ValueError("unable to load issue state validator") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.validate_state(state, repository) + + def reference_values(primitive: dict[str, Any], ref_type: str) -> set[str]: refs = primitive.get("refs") if not isinstance(refs, list): @@ -56,7 +74,15 @@ def merge_evidence( if ( isinstance(catalog_entry, dict) and catalog_entry.get("kind") == item.get("kind") + and catalog_entry.get("signatureVersion") == SIGNATURE_VERSION and isinstance(catalog_entry.get("signature"), dict) + and stable_hash( + { + "kind": item.get("kind"), + "signature": catalog_entry["signature"], + } + ) + == item.get("fingerprint") ): restored["signature"] = catalog_entry["signature"] evidence_by_key[str(item["evidenceKey"])] = restored @@ -97,7 +123,9 @@ def build_fingerprint_catalog( if signature_bytes > MAX_SIGNATURE_BYTES: continue current = entries.get(fingerprint) - if current is None or completed_at > current["lastSeenAt"]: + if current is None or parse_timestamp(completed_at) > parse_timestamp( + current["lastSeenAt"] + ): entries[fingerprint] = { "kind": kind, "signatureVersion": SIGNATURE_VERSION, @@ -428,8 +456,9 @@ def main() -> None: "proposalQueue": proposal_queue if isinstance(proposal_queue, list) else [], "proposalHistory": proposal_history, } + validated_state = validate_next_state(next_state, args.repository) write_json(args.output, result) - write_json(args.state_out, next_state) + write_json(args.state_out, validated_state) if __name__ == "__main__": diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py index 43f0800..2a93d31 100644 --- a/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py +++ b/plugins/repo-dreamer/skills/repository-skill-forge/scripts/issue-state.py @@ -260,7 +260,10 @@ def validate_fingerprint_catalog(catalog: Any) -> None: not isinstance(tokens, list) or not tokens or len(tokens) > 40 - or any(not isinstance(token, str) or len(token) > 80 for token in tokens) + or any( + not isinstance(token, str) or not token or len(token) > 80 + for token in tokens + ) ): raise ValueError( "issue state command fingerprint has invalid tokens" diff --git a/plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py b/plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py index 3c1fcd7..c2e347b 100644 --- a/plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py +++ b/plugins/repo-dreamer/skills/repository-skill-forge/tests/test_state_catalog.py @@ -108,6 +108,46 @@ def test_catalog_restores_historical_signature_for_aggregation(self) -> None: self.assertEqual(signature, patterns[0]["signature"]) + def test_catalog_does_not_restore_inconsistent_signature(self) -> None: + signature = {"tokens": ["git", "show", ""]} + fingerprint = stable_hash({"kind": "command", "signature": signature}) + state = empty_state() | { + "observations": [observation(fingerprint)], + "fingerprintCatalog": { + fingerprint: { + "kind": "command", + "signatureVersion": 2, + "signature": {"tokens": ["git", "status"]}, + "lastSeenAt": "2026-08-13T00:00:00Z", + } + }, + } + + evidence, _history = aggregate.merge_evidence( + state, + {"primitives": []}, + "owner/repository", + ) + + self.assertNotIn("signature", evidence[0]) + + def test_catalog_rejects_empty_command_token(self) -> None: + signature = {"tokens": ["git", ""]} + fingerprint = stable_hash({"kind": "command", "signature": signature}) + state = empty_state() | { + "fingerprintCatalog": { + fingerprint: { + "kind": "command", + "signatureVersion": 1, + "signature": signature, + "lastSeenAt": "2026-08-13T00:00:00Z", + } + } + } + + with self.assertRaisesRegex(ValueError, "invalid tokens"): + issue_state.validate_state(state, "owner/repository") + def test_catalog_rejects_secret_shaped_signature_values(self) -> None: signature = {"tokens": ["deploy", "token=abcdefghijklmnop"]} fingerprint = stable_hash({"kind": "command", "signature": signature}) @@ -167,6 +207,48 @@ def test_catalog_builder_enforces_entry_and_byte_limits(self) -> None: aggregate.MAX_FINGERPRINT_CATALOG_BYTES, ) + def test_catalog_builder_compares_parsed_timestamps(self) -> None: + signature = {"tokens": ["git", "status"]} + fingerprint = stable_hash({"kind": "command", "signature": signature}) + catalog = aggregate.build_fingerprint_catalog( + [ + { + "fingerprint": fingerprint, + "kind": "command", + "signature": signature, + "completedAt": "2026-08-13T10:00:00+02:00", + }, + { + "fingerprint": fingerprint, + "kind": "command", + "signature": signature, + "completedAt": "2026-08-13T09:00:00Z", + }, + ] + ) + + self.assertEqual( + "2026-08-13T09:00:00Z", + catalog[fingerprint]["lastSeenAt"], + ) + + def test_next_state_is_validated_before_persistence(self) -> None: + signature = {"tokens": ["deploy", "token=abcdefghijklmnop"]} + fingerprint = stable_hash({"kind": "command", "signature": signature}) + state = empty_state() | { + "fingerprintCatalog": { + fingerprint: { + "kind": "command", + "signatureVersion": 1, + "signature": signature, + "lastSeenAt": "2026-08-13T00:00:00Z", + } + } + } + + with self.assertRaisesRegex(ValueError, "secret-shaped"): + aggregate.validate_next_state(state, "owner/repository") + if __name__ == "__main__": unittest.main()