tegwick
a18d83f74a
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e738-9e15-7600-87e6-90dec3a7beef
77 lines
4.4 KiB
Diff
77 lines
4.4 KiB
Diff
--- a/src/activity_core/context_resolvers/frontend_patterns.py
|
|
+++ b/src/activity_core/context_resolvers/frontend_patterns.py
|
|
@@ -10,6 +10,7 @@
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
+import math
|
|
from pathlib import PurePosixPath
|
|
import re
|
|
import time
|
|
@@ -149,6 +150,44 @@
|
|
return latest
|
|
|
|
|
|
+def monthly_review(root_sha: str, now: datetime, start: datetime, end: datetime) -> dict:
|
|
+ """Bounded, source-pinned owner snapshot and measured research totals.
|
|
+
|
|
+ This is review input, never authority or a live health attestation.
|
|
+ Missing/invalid input degrades the monthly packet while preserving receipts.
|
|
+ """
|
|
+ data = json.loads(raw("frontend-patterns", root_sha, "feedback/monthly-review-input.json"))
|
|
+ require(data["schema_version"] == 1, "invalid_monthly_input")
|
|
+ stamp = datetime.fromisoformat(data["as_of"].replace("Z", "+00:00"))
|
|
+ require(stamp.tzinfo is not None and stamp <= now, "invalid_monthly_date")
|
|
+ health = data["health"]
|
|
+ keys = {"schedule", "collection", "verification", "publication", "rollback", "executor"}
|
|
+ require(isinstance(health, dict) and set(health) == keys, "invalid_monthly_health")
|
|
+ require(all(isinstance(v, str) and 0 < len(v) <= 80 for v in health.values()), "invalid_monthly_health")
|
|
+ require(path_ok(data["evidence_ref"]), "invalid_monthly_evidence")
|
|
+ # The cited file must exist at this same source revision.
|
|
+ raw("frontend-patterns", root_sha, data["evidence_ref"])
|
|
+ decisions = data["decisions"]
|
|
+ require(isinstance(decisions, list) and len(decisions) <= 3, "invalid_monthly_decisions")
|
|
+ for decision in decisions:
|
|
+ require(isinstance(decision, dict) and set(decision) == {"question", "recommendation", "default"}, "invalid_monthly_decision")
|
|
+ require(all(isinstance(v, str) and 0 < len(v) <= 100 for v in decision.values()), "invalid_monthly_decision")
|
|
+ ledger = json.loads(raw("frontend-patterns", root_sha, "research/evolution/ledger.json"))
|
|
+ events = ledger["events"]
|
|
+ require(isinstance(events, list) and len(events) <= 1000, "invalid_learning_ledger")
|
|
+ selected = [e for e in events if start.date().isoformat() <= e["date"] < end.date().isoformat()]
|
|
+ totals = {}
|
|
+ for field in ("research_minutes", "execution_minutes", "reported_tokens", "cost_usd"):
|
|
+ values = [e["effort"][field] for e in selected]
|
|
+ require(all(v is None or (type(v) in (int, float) and math.isfinite(v) and v >= 0) for v in values), "invalid_effort")
|
|
+ totals[field] = {"known": round(sum(v for v in values if v is not None), 3), "unknown_cycles": sum(v is None for v in values)}
|
|
+ return {"basis": "owner snapshot; not live attestation", "as_of": data["as_of"],
|
|
+ "stale": (now - stamp).total_seconds() > 7 * 86400,
|
|
+ "health": health, "evidence_ref": data["evidence_ref"], "decisions": decisions,
|
|
+ "default": "Retain current policy/budgets; no new authority or maturity claim.",
|
|
+ "research_cycles": len(selected), "measured_effort": totals}
|
|
+
|
|
+
|
|
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)
|
|
@@ -165,6 +204,7 @@
|
|
"workplan": "FEP-WP-0008", "executor_dependency": "GLAS-WP-0012 / HFACT-WP-0001",
|
|
}
|
|
config = {"sources": []}
|
|
+ root_sha = None
|
|
rows = []
|
|
events_seen = set()
|
|
total = 0
|
|
@@ -213,6 +253,13 @@
|
|
before = first - timedelta(days=1)
|
|
start = before.replace(day=1)
|
|
report["window"] = {"start": start.isoformat(), "end_exclusive": first.isoformat()}
|
|
+ try:
|
|
+ require(root_sha is not None, "missing_root_revision")
|
|
+ report["review"] = monthly_review(root_sha, now, start, first)
|
|
+ except (ValueError, KeyError, TypeError, AttributeError, OverflowError, httpx.HTTPError):
|
|
+ report["review"] = {"status": "unavailable", "health": "unknown", "decisions": [],
|
|
+ "default": "Retain current policy/budgets; no new authority or maturity claim."}
|
|
+ report["errors"].append("monthly_review_input_invalid_or_unavailable")
|
|
# 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"])
|