Add bounded frontend-patterns feedback reporting resolver

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e241-8285-7a63-8c0c-51c9cb824dc3

src/activity_core/context_resolvers/__init__.py

@@ -1,5 +1,6 @@
 from activity_core.context_resolvers import (  # noqa: F401
     core_hub,
+    frontend_patterns,
     event_payload,
     kaizen,
     ops_inventory,

src/activity_core/context_resolvers/frontend_patterns.py

@@ -0,0 +1,254 @@
+"""Read-only, bounded collection for the frontend-patterns reference loop.
+
+Returns a compact JSON digest for the existing deterministic report instruction.
+Never executes feedback, invokes a model, or authorizes repository mutation.
+"""
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+from contextvars import ContextVar
+from datetime import datetime, timezone
+import hashlib
+import json
+from pathlib import PurePosixPath
+import re
+import time
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import httpx
+import yaml
+
+from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver
+
+HOST = "https://forgejo.coulomb.social"
+TOPIC = "fe2aaa78-9c20-4feb-b3d2-4fe0529572a3"
+SHA = re.compile(r"[0-9a-f]{40}")
+DIGEST = re.compile(r"[0-9a-f]{64}")
+SLUG = re.compile(r"[a-z0-9][a-z0-9-]{0,79}")
+ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9:_.-]{0,159}")
+MAX_BYTES = 131072
+MAX_RECEIPTS = 100
+DEADLINE: ContextVar[float | None] = ContextVar("frontend_patterns_deadline", default=None)
+
+
+def require(ok: Any, message: str) -> None:
+    if not ok:
+        raise ValueError(message)
+
+
+def path_ok(value: Any) -> bool:
+    return (isinstance(value, str) and bool(value) and len(value) <= 240
+            and not PurePosixPath(value).is_absolute()
+            and all(part not in {"", ".", ".."} for part in value.split("/"))
+            and re.fullmatch(r"[A-Za-z0-9_./-]+", value) is not None)
+
+
+def fetch(url: str) -> bytes:
+    # Callers construct only fixed-host URLs from validated slug/SHA/path parts.
+    deadline = DEADLINE.get()
+    remaining = deadline - time.monotonic() if deadline is not None else 15
+    require(remaining > 0, "collection_time_limit")
+    with httpx.stream("GET", url, timeout=min(5, remaining), follow_redirects=False) as response:
+        response.raise_for_status()
+        body = bytearray()
+        for part in response.iter_bytes():
+            require(deadline is None or time.monotonic() < deadline, "collection_time_limit")
+            body.extend(part)
+            require(len(body) <= MAX_BYTES, "response_size_limit")
+        return bytes(body)
+
+
+def revision(repo: str, branch: str) -> str:
+    require(SLUG.fullmatch(repo) and SLUG.fullmatch(branch), "invalid_repository_or_branch")
+    data = json.loads(fetch(f"{HOST}/api/v1/repos/coulomb/{repo}/branches/{branch}"))
+    sha = data["commit"]["id"]
+    require(isinstance(sha, str) and SHA.fullmatch(sha), "invalid_revision")
+    return sha
+
+
+def raw(repo: str, sha: str, path: str) -> bytes:
+    require(SLUG.fullmatch(repo) and SHA.fullmatch(sha) and path_ok(path), "invalid_source_path")
+    return fetch(f"{HOST}/coulomb/{repo}/raw/commit/{sha}/{path}")
+
+
+def validate_event(event: Any, repo: str) -> None:
+    require(isinstance(event, dict) and event.get("schema_version") == 1, "invalid_event_schema")
+    for field in ("use_id", "event_id"):
+        require(isinstance(event.get(field), str) and ID.fullmatch(event[field]), "invalid_identity")
+    require(event.get("kind") in {"real", "synthetic"}, "invalid_kind")
+    previous = event.get("previous_event_sha256")
+    require(previous is None or isinstance(previous, str) and DIGEST.fullmatch(previous), "invalid_predecessor")
+    stamp = datetime.fromisoformat(event["recorded_at"].replace("Z", "+00:00"))
+    require(stamp.tzinfo is not None, "timestamp_requires_timezone")
+    require(stamp <= datetime.now(timezone.utc), "future_event")
+    consumer = event["consumer"]
+    require(consumer["repo"] == repo and SHA.fullmatch(consumer["revision"]), "consumer_mismatch")
+    require(all(isinstance(consumer[k], str) and consumer[k] for k in ("task_ref", "producer", "run_ref")), "missing_provenance")
+    reference = event["reference"]
+    require(SHA.fullmatch(reference["packaging_revision"]) and DIGEST.fullmatch(reference["manifest_sha256"]), "invalid_reference_pin")
+    require(isinstance(reference["version"], str) and reference["version"], "missing_reference_version")
+    context = event["context"]
+    require(isinstance(context["problem"], str) and 0 < len(context["problem"]) <= 1000, "invalid_problem")
+    require(isinstance(context["constraints"], list) and all(isinstance(x, str) and len(x) <= 500 for x in context["constraints"]), "invalid_constraints")
+    for field in ("considered", "selected"):
+        values = context[field]
+        require(isinstance(values, list) and len(values) <= 17 and len(set(values)) == len(values), "invalid_candidates")
+        require(all(isinstance(x, str) and re.fullmatch(r"CAND-\d{4}", x) and 1 <= int(x[5:]) <= 17 for x in values), "unknown_candidate")
+    require(set(context["selected"]) <= set(context["considered"]), "unconsidered_selection")
+    action = event["decision"]["action"]
+    require(action in {"applied", "adapted", "rejected", "no-match"}, "invalid_action")
+    require(bool(context["selected"]) == (action in {"applied", "adapted"}), "action_selection_mismatch")
+    require(isinstance(event["decision"]["reason"], str) and event["decision"]["reason"], "missing_decision_reason")
+    outcome = event["outcome"]
+    require(outcome["value"] in {"helpful", "harmful", "mixed", "no-effect", "unknown"}, "invalid_outcome")
+    require(outcome["basis"] in {"consumer-report", "model-assessment", "observed-result"}, "invalid_basis")
+    require(outcome["stage"] in {"decision", "implementation", "test", "operation"}, "invalid_stage")
+    require(isinstance(outcome["expectation"], str) and outcome["expectation"], "missing_expectation")
+    refs = outcome["evidence_refs"]
+    require(isinstance(refs, list) and len(refs) <= 10 and all(isinstance(x, str) and 0 < len(x) <= 1000 for x in refs), "invalid_evidence")
+    require(outcome["value"] == "unknown" or bool(refs) and isinstance(outcome["observation"], str) and bool(outcome["observation"]), "unevidenced_outcome")
+
+
+def reduce_events(rows: list[tuple[dict, str]], errors: list[str]) -> list[dict]:
+    # A use is quarantined as a whole on identity collision or ambiguous lineage.
+    by_use: dict[str, dict[str, dict]] = defaultdict(dict)
+    identities: dict[tuple[str, str], tuple[str, str]] = {}
+    bad = set()
+    for event, digest in rows:
+        use = event["use_id"]
+        key = (event["consumer"]["repo"], event["event_id"])
+        if key in identities and identities[key] != (use, digest):
+            bad.update((use, identities[key][0]))
+        identities[key] = (use, digest)
+        by_use[use][digest] = event
+    latest = []
+    for use, nodes in by_use.items():
+        roots = [d for d, e in nodes.items() if e["previous_event_sha256"] is None]
+        seen = set()
+        current = roots[0] if len(roots) == 1 else None
+        previous_event = None
+        while current is not None and current not in seen:
+            seen.add(current)
+            event = nodes[current]
+            if previous_event is not None:
+                same = all(event[k] == previous_event[k] for k in ("kind", "reference", "context", "decision"))
+                same = same and event["consumer"]["repo"] == previous_event["consumer"]["repo"]
+                newer = datetime.fromisoformat(event["recorded_at"].replace("Z", "+00:00")) >= datetime.fromisoformat(previous_event["recorded_at"].replace("Z", "+00:00"))
+                if not same or not newer:
+                    bad.add(use)
+            children = [d for d, e in nodes.items() if e["previous_event_sha256"] == current]
+            if len(children) > 1:
+                bad.add(use)
+            previous_event = event
+            current = children[0] if len(children) == 1 else None
+        if len(seen) != len(nodes) or use in bad:
+            errors.append("quarantined_event_lineage")
+        elif previous_event is not None:
+            latest.append(previous_event)
+    return latest
+
+
+def collect(mode: str = "daily", now: datetime | None = None) -> dict:
+    require(mode in {"daily", "weekly", "monthly"}, "invalid_mode")
+    now = now or datetime.now(timezone.utc)
+    report: dict[str, Any] = {
+        "repo": "frontend-patterns", "mode": mode, "generated_at": now.isoformat(),
+        "status": "ok", "execution_ready": False, "model_calls": 0,
+        "eligible_tasks": None, "instrumented_repositories": 0,
+        "source_revisions": [], "receipt_events": 0, "duplicates": 0,
+        "real_uses": 0, "synthetic_uses": 0, "outcomes": {}, "decisions": {},
+        "signals": [], "errors": [],
+        "limits": ["Producer outcomes are not independently verified.",
+                   "No admitted budget-enforced repository executor; automatic edits disabled.",
+                   "Missing consumer instrumentation means coverage is unknown."],
+        "workplan": "FEP-WP-0008", "executor_dependency": "GLAS-WP-0012 / HFACT-WP-0001",
+    }
+    config = {"sources": []}
+    rows = []
+    events_seen = set()
+    total = 0
+    try:
+        root_sha = revision("frontend-patterns", "main")
+        report["source_revisions"].append("frontend-patterns@" + root_sha)
+        config = json.loads(raw("frontend-patterns", root_sha, "feedback/sources.json"))
+        require(config["schema_version"] == 1 and isinstance(config["sources"], list) and len(config["sources"]) <= 10, "invalid_source_roster")
+        source_ids = [s["repo"] for s in config["sources"]]
+        require(len(set(source_ids)) == len(source_ids), "duplicate_source")
+        for source in config["sources"]:
+            repo = source["repo"]
+            try:
+                sha = revision(repo, source["branch"])
+                report["source_revisions"].append(repo + "@" + sha)
+                index = json.loads(raw(repo, sha, source["index"]))
+                require(index["schema_version"] == 1 and isinstance(index["receipts"], list), "invalid_receipt_index")
+                total += len(index["receipts"])
+                require(total <= MAX_RECEIPTS, "collection_limit_exceeded")
+                report["instrumented_repositories"] += 1
+                for item in index["receipts"]:
+                    try:
+                        require(isinstance(item["sha256"], str) and DIGEST.fullmatch(item["sha256"]), "invalid_digest")
+                        data = raw(repo, sha, item["path"])
+                        digest = hashlib.sha256(data).hexdigest()
+                        require(digest == item["sha256"], "receipt_digest_mismatch")
+                        event = yaml.safe_load(data)
+                        validate_event(event, repo)
+                        if digest in events_seen:
+                            report["duplicates"] += 1
+                            continue
+                        events_seen.add(digest)
+                        rows.append((event, digest))
+                    except (ValueError, KeyError, TypeError, AttributeError, yaml.YAMLError, httpx.HTTPError):
+                        report["errors"].append("receipt_invalid_or_unavailable")
+            except (ValueError, KeyError, TypeError, AttributeError, httpx.HTTPError):
+                report["errors"].append("consumer_source_invalid_or_unavailable")
+    except (ValueError, KeyError, TypeError, AttributeError, httpx.HTTPError):
+        report["errors"].append("roster_invalid_or_unavailable")
+    report["receipt_events"] = len(rows)
+    latest = reduce_events(rows, report["errors"])
+    local = now.astimezone(ZoneInfo("Europe/Berlin"))
+    first = local.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
+    if mode == "monthly":
+        from datetime import timedelta
+        before = first - timedelta(days=1)
+        start = before.replace(day=1)
+        report["window"] = {"start": start.isoformat(), "end_exclusive": first.isoformat()}
+        # Select latest event as of the end of the month, not latest today.
+        eligible_rows = [(e, d) for e, d in rows if datetime.fromisoformat(e["recorded_at"].replace("Z", "+00:00")) < first]
+        latest = reduce_events(eligible_rows, report["errors"])
+        latest = [e for e in latest if datetime.fromisoformat(e["recorded_at"].replace("Z", "+00:00")) >= start]
+    report["synthetic_uses"] = sum(e["kind"] == "synthetic" for e in latest)
+    real = [e for e in latest if e["kind"] == "real"]
+    report["real_uses"] = len(real)
+    report["outcomes"] = dict(Counter(e["outcome"]["value"] for e in real))
+    report["decisions"] = dict(Counter(e["decision"]["action"] for e in real))
+    signals = []
+    for e in real:
+        value = e["outcome"]["value"]
+        if value in {"harmful", "mixed", "helpful"}:
+            signals.append({"use_id": e["use_id"], "outcome": value, "verification_required": True})
+    signals.sort(key=lambda s: ({"harmful": 0, "mixed": 1, "helpful": 2}[s["outcome"]], s["use_id"]))
+    report["signals"] = signals[:3]
+    report["unselected_signals"] = max(0, len(signals) - 3)
+    report["errors"] = dict(Counter(report["errors"]))
+    if report["errors"]:
+        report["status"] = "degraded"
+    elif not config["sources"]:
+        report["status"] = "no_consumers"
+    report["summary"] = f"Frontend patterns {mode}: {report['status']}; {len(real)} real uses, {len(signals)} unverified outcome signals. Automatic edits disabled."
+    return report
+
+
+class FrontendPatternsResolver(ContextResolver):
+    def resolve(self, query: str, event: Any, params: dict[str, Any]) -> str:
+        token = DEADLINE.set(time.monotonic() + 45)
+        try:
+            report = collect(query)
+        finally:
+            DEADLINE.reset(token)
+        encoded = json.dumps(report, sort_keys=True, separators=(",", ":"))
+        require(len(encoded) <= 3900, "digest_size_limit")
+        return encoded
+
+
+CONTEXT_RESOLVER_REGISTRY["frontend-patterns"] = FrontendPatternsResolver