a8caf27c14
Five weeks of forge evidence per week rather than a rolling window, so a finding survives under one key and "closed after three weeks" is computable. Rules run deterministically; agent.py hands the packet to whoever writes the prose and never calls a model itself. Model choice fails closed — no default, no substitute, and the exact named model must be served or the pass refuses to run.
353 lines
16 KiB
Python
353 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""The agent's door into project state. No inference happens here, ever.
|
|
|
|
The deterministic layer establishes what is true; the agent reading this writes
|
|
the prose. Nothing in this file selects, calls, or configures a model — the
|
|
agent invoking it is the model, and it already has the identity, the context and
|
|
the tools.
|
|
|
|
agent.py status what is collected, what is missing
|
|
agent.py packet --week 2026-W33 everything needed to write the draft
|
|
agent.py explain f-0069 one finding and every row it rests on
|
|
agent.py draft --week 2026-W33 --file draft.md --author Annie
|
|
agent.py verdict f-0069 promote --note "worth a task"
|
|
|
|
`packet` is the contract. It carries the week's state, what turned, what has
|
|
stood longest, and an `investigate` list naming where the deterministic layer
|
|
knows it is thin — go read those repositories rather than trusting the packet.
|
|
|
|
See docs/tasks/79-project-state.md.
|
|
"""
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import common as c
|
|
import collect
|
|
import synthesize as s
|
|
|
|
STORE = s.STORE
|
|
CONTRACT = {
|
|
"you_are": "the agent that writes Souveraine's weekly project-state draft",
|
|
"the_split": (
|
|
"Everything under `counted`, `movement`, `standing`, `carried` and `closed` was "
|
|
"computed from the forge by deterministic code and is authoritative. Your draft is "
|
|
"downstream of it and is disposable — it can be regenerated from this packet and "
|
|
"the stored state at any time."
|
|
),
|
|
"produce": (
|
|
"A markdown draft answering: what became true about Souveraine this week, what "
|
|
"stopped being true, what advanced, what regressed, what remains unresolved, what "
|
|
"decisions are waiting, and which milestones were crossed or are approaching."
|
|
),
|
|
"rules": [
|
|
"Cite finding ids (f-0069) and evidence ids (c:, ci:, d:, i:, p:, r:) for every claim.",
|
|
"Do not invent a shipment, release, decision, or outcome that is not in the evidence.",
|
|
"'Nothing moved here this week' is a legal sentence.",
|
|
"Where the packet's `investigate` list says the evidence is thin, go read the "
|
|
"repository before asserting; say so inline if you did not.",
|
|
"The report is internal. It publishes nowhere until a human promotes it.",
|
|
],
|
|
"when_the_packet_is_not_enough": (
|
|
"Read the repositories directly. `evidence.file` holds every row including full "
|
|
"commit bodies and changed paths; the working trees named in `repos` are checked "
|
|
"out locally. `agent.py explain <id>` prints any finding with its rows in full."
|
|
),
|
|
"return_with": "agent.py draft --week <week> --file <path> --author <you> [--model <id>]",
|
|
"no_inference_here": (
|
|
"This tool never calls a model. Model choice belongs to whoever invokes it and is "
|
|
"recorded, not decided, by --model."
|
|
),
|
|
}
|
|
|
|
EVIDENCE_KINDS = {
|
|
"c:": "commit — repo, sha, subject, body, changed paths, dimension and the basis that chose it",
|
|
"ci:": "Gitea Actions run — workflow, branch, head sha, conclusion; joins to a commit id",
|
|
"d:": "declared change — a task-index row or a SAF/docs file that moved",
|
|
"i:": "issue", "p:": "pull request", "r:": "release", "w:": "the week's rollup",
|
|
}
|
|
|
|
|
|
def load_state():
|
|
return s.load(os.path.join(STORE, "state.json"), {"findings": []})
|
|
|
|
|
|
def load_evidence(week):
|
|
return s.load(os.path.join(STORE, "evidence", f"{week}.json"))
|
|
|
|
|
|
def status():
|
|
state = load_state()
|
|
held = collect.held()
|
|
return {
|
|
"weeks_held": held,
|
|
"latest_synthesized": state.get("latest_week"),
|
|
"gaps": collect.gaps(),
|
|
"pending": s.pending(state),
|
|
"open_findings": sum(1 for f in state.get("findings", []) if f["status"] == "open"),
|
|
"drafts": sorted(os.listdir(os.path.join(STORE, "drafts")))
|
|
if os.path.isdir(os.path.join(STORE, "drafts")) else [],
|
|
"next": next_action(state, held),
|
|
}
|
|
|
|
|
|
def next_action(state, held):
|
|
if collect.gaps():
|
|
return "collect.py --backfill — weeks are missing evidence"
|
|
if s.pending(state):
|
|
return "synthesize.py --catch-up — weeks are collected but not passed over"
|
|
if not held:
|
|
return "collect.py --backfill --since <week> — nothing collected yet"
|
|
return f"agent.py packet --week {state.get('latest_week')} — state is current; write the draft"
|
|
|
|
|
|
def thin_spots(ev, findings):
|
|
"""Where the deterministic layer knows it did not see enough.
|
|
|
|
This is the half of the packet that asks for work: each entry says what is
|
|
weak, and what reading it would take to settle.
|
|
"""
|
|
out = []
|
|
unplaced = [r for r in ev["commits"] if r["dimension"] == c.UNASSIGNED]
|
|
if unplaced:
|
|
out.append({
|
|
"why": f"{len(unplaced)} commit(s) matched no dimension keyword and were not guessed at",
|
|
"how": "read these commits in their repositories and place them yourself",
|
|
"rows": [{"id": r["id"], "repo": r["repo"], "subject": r["subject"],
|
|
"paths": r["paths"][:6]} for r in unplaced],
|
|
})
|
|
blind = [name for name, meta in ev["collector"]["repos"].items()
|
|
if not meta["checkout"] and meta["commits"]]
|
|
if blind:
|
|
out.append({
|
|
"why": f"no local checkout for {', '.join(blind)}, so changed paths were unavailable "
|
|
f"and those commits were placed by subject or repository alone",
|
|
"how": "clone or fetch those repos if a dimension call there matters this week",
|
|
"rows": [],
|
|
})
|
|
single = [f for f in findings if len(f["evidence"]) == 1
|
|
and f["provenance"]["rule"] not in ("task-standing", "pull-aging", "issue-aging")]
|
|
if single:
|
|
out.append({
|
|
"why": f"{len(single)} finding(s) rest on exactly one evidence row",
|
|
"how": "agent.py explain <id>, then read the surrounding work before leaning on it",
|
|
"rows": [{"id": f["id"], "title": f["title"], "evidence": f["evidence"]}
|
|
for f in single[:12]],
|
|
})
|
|
declared_only = [dim for dim, r in ev["rollup"]["by_dimension"].items()
|
|
if r.get("declared") and not r.get("commits")]
|
|
if declared_only:
|
|
out.append({
|
|
"why": f"{', '.join(declared_only)} changed in the documentation with no commits behind it",
|
|
"how": "a doc can lead the code or lag it — check which, before calling it advanced",
|
|
"rows": [{"id": r["id"], "file": r.get("file"), "title": r.get("title"),
|
|
"change": r["change"]}
|
|
for r in ev["declared"] if r["dimension"] in declared_only][:12],
|
|
})
|
|
notes = ev["collector"].get("notes") or []
|
|
if notes:
|
|
out.append({"why": "the collector reported problems", "how": "treat these areas as incomplete",
|
|
"rows": [{"note": n} for n in notes]})
|
|
return out
|
|
|
|
|
|
def packet(week, brief=False):
|
|
label, monday, sunday = c.parse_week(week)
|
|
ev = load_evidence(label)
|
|
if not ev:
|
|
return None, f"no evidence for {label} — collect.py --week {label}"
|
|
state = s.load(os.path.join(STORE, "history", f"{label}.json")) or load_state()
|
|
if label not in state.get("weeks", []):
|
|
return None, f"{label} is collected but not synthesized — synthesize.py --catch-up"
|
|
|
|
findings = state["findings"]
|
|
this_week = [f for f in findings if f["last_seen"] == label and f["status"] == "open"]
|
|
quiet = [f for f in this_week if f["provenance"]["rule"] == "task-standing"
|
|
and f["movement"] != "blocked"]
|
|
moved = [f for f in this_week if f not in quiet]
|
|
carried = [f for f in findings if f["status"] == "open" and f["last_seen"] != label]
|
|
closed = [f for f in findings if f.get("resolved_in") == label]
|
|
turned = [f for f in moved if f.get("was") and f["was"]["movement"] != f["movement"]]
|
|
|
|
def slim(f):
|
|
row = {"id": f["id"], "key": f["key"], "dimension": f["dimension"],
|
|
"movement": f["movement"], "title": f["title"], "statement": f["statement"],
|
|
"evidence": f["evidence"], "rule": f["provenance"]["rule"],
|
|
"first_seen": f["first_seen"], "weeks_open": f.get("weeks_open")}
|
|
for extra in ("was", "weeks_in_section", "section_since", "task_state", "resolved_in"):
|
|
if f.get(extra):
|
|
row[extra] = f[extra]
|
|
if (f.get("human") or {}).get("verdict"):
|
|
row["human"] = f["human"]
|
|
return row
|
|
|
|
commits = [{"id": r["id"], "repo": r["repo"], "at": r["at"], "dimension": r["dimension"],
|
|
"basis": r["basis"], "subject": r["subject"],
|
|
**({} if brief else {"detail": r["detail"], "paths": r["paths"][:12]})}
|
|
for r in ev["commits"]]
|
|
# Only a finding that carries can be the oldest one; a CI week or a
|
|
# dimension's activity describes one week and lapses with it.
|
|
oldest = max((f for f in findings if f["status"] == "open"
|
|
and f["provenance"]["rule"] in s.CARRIES),
|
|
key=lambda f: (f.get("weeks_open", 1), f["id"]), default=None)
|
|
|
|
return {
|
|
"contract": CONTRACT,
|
|
"week": label,
|
|
"range": ev["range"],
|
|
"completeness": {
|
|
"weeks_held": collect.held(),
|
|
"gaps": collect.gaps(),
|
|
"pending": s.pending(load_state()),
|
|
"collected_at": ev["collected_at"],
|
|
"collector": {k: ev["collector"][k] for k in ("tool", "version", "forge", "api_calls",
|
|
"redacted")},
|
|
},
|
|
"counted": ev["rollup"],
|
|
"movement": [slim(f) for f in moved],
|
|
"standing": [slim(f) for f in sorted(quiet, key=lambda f: -f.get("weeks_in_section", 1))],
|
|
"carried": [slim(f) for f in sorted(carried, key=lambda f: -f.get("weeks_open", 1))],
|
|
"closed": [slim(f) for f in closed],
|
|
"continuity": {
|
|
"weeks_available": state.get("weeks", []),
|
|
"turned_this_week": [{"id": f["id"], "title": f["title"],
|
|
"from": f["was"]["movement"], "to": f["movement"],
|
|
"since": f["was"]["week"]} for f in turned],
|
|
"oldest_open": slim(oldest) if oldest else None,
|
|
},
|
|
"investigate": thin_spots(ev, this_week),
|
|
# --brief keeps every row the counts cannot replace and drops the rest:
|
|
# green CI runs are already a tally, and the task index is `standing`.
|
|
"evidence": {
|
|
"file": f"project-state/evidence/{label}.json",
|
|
"explain": "tools/project-state/agent.py explain <finding-id>",
|
|
"kinds": EVIDENCE_KINDS,
|
|
"commits": commits,
|
|
"declared": ev["declared"],
|
|
"ci": [r for r in ev["ci"] if r["conclusion"] != "success"] if brief else ev["ci"],
|
|
"ci_note": "green runs omitted; counts are under `counted.ci`" if brief else None,
|
|
"issues": ev["issues"],
|
|
"pulls": ev["pulls"],
|
|
"releases": ev["releases"],
|
|
**({} if brief else {"task_index": ev.get("index", {})}),
|
|
},
|
|
"repos": {name: {"commits": meta["commits"],
|
|
"checkout": os.path.join(c.PROJECTS, name) if meta["checkout"] else None}
|
|
for name, meta in ev["collector"]["repos"].items() if meta["commits"]},
|
|
}, None
|
|
|
|
|
|
def record_draft(week, path, author, model, note):
|
|
label, _, _ = c.parse_week(week)
|
|
try:
|
|
body = open(path).read()
|
|
except OSError as e:
|
|
note(f"cannot read {path}: {e}")
|
|
return 2
|
|
state = s.load(os.path.join(STORE, "history", f"{label}.json")) or load_state()
|
|
folder = os.path.join(STORE, "drafts")
|
|
os.makedirs(folder, exist_ok=True)
|
|
slug = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in author).strip("-").lower()
|
|
stem = os.path.join(folder, f"{label}--{slug}")
|
|
open(stem + ".md", "w").write(body)
|
|
# The draft names the state it was written against, so a later reader can
|
|
# tell whether it still describes the findings that exist.
|
|
open(stem + ".json", "w").write(json.dumps({
|
|
"week": label,
|
|
"author": author,
|
|
"model": model,
|
|
"recorded_at": c.iso_z(dt.datetime.now(dt.timezone.utc)),
|
|
"state_updated": state.get("updated"),
|
|
"findings_open": sum(1 for f in state.get("findings", []) if f["status"] == "open"),
|
|
"evidence": f"project-state/evidence/{label}.json",
|
|
"published": False,
|
|
"source": os.path.abspath(path),
|
|
}, indent=2, ensure_ascii=False) + "\n")
|
|
note(f"{label}: draft by {author}"
|
|
+ (f" ({model})" if model else "")
|
|
+ f" -> project-state/drafts/{label}--{slug}.md; nothing published")
|
|
return 0
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
|
|
sub = ap.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("status", help="what is collected, synthesized, and missing")
|
|
|
|
p = sub.add_parser("packet", help="everything needed to write the week's draft")
|
|
p.add_argument("--week", default="last")
|
|
p.add_argument("--brief", action="store_true", help="drop commit bodies and paths")
|
|
p.add_argument("--out", help="write here instead of stdout")
|
|
|
|
e = sub.add_parser("explain", help="one finding and every evidence row it cites")
|
|
e.add_argument("finding")
|
|
e.add_argument("--json", action="store_true")
|
|
|
|
d = sub.add_parser("draft", help="record a draft written by an agent or a person")
|
|
d.add_argument("--week", default="last")
|
|
d.add_argument("--file", required=True)
|
|
d.add_argument("--author", required=True)
|
|
d.add_argument("--model", help="the model that wrote it, recorded not chosen")
|
|
|
|
v = sub.add_parser("verdict", help="a person's word on a finding")
|
|
v.add_argument("finding")
|
|
v.add_argument("verdict", choices=s.VERDICTS)
|
|
v.add_argument("--note")
|
|
v.add_argument("--by")
|
|
|
|
args = ap.parse_args()
|
|
|
|
def note(message):
|
|
print(f"[project-state] {message}", file=sys.stderr)
|
|
|
|
if args.command == "status":
|
|
print(json.dumps(status(), indent=2, ensure_ascii=False))
|
|
return 0
|
|
|
|
if args.command == "packet":
|
|
data, problem = packet(args.week, args.brief)
|
|
if problem:
|
|
note(problem)
|
|
return 2
|
|
text = json.dumps(data, indent=2, ensure_ascii=False) + "\n"
|
|
if args.out:
|
|
open(args.out, "w").write(text)
|
|
note(f"{data['week']}: packet -> {args.out} ({len(text):,} chars)")
|
|
else:
|
|
sys.stdout.write(text)
|
|
return 0
|
|
|
|
if args.command == "explain":
|
|
if not args.json:
|
|
return s.explain(load_state(), args.finding)
|
|
state = load_state()
|
|
f = next((x for x in state["findings"]
|
|
if x["id"] == args.finding or x["key"] == args.finding), None)
|
|
if not f:
|
|
note(f"no finding {args.finding}")
|
|
return 1
|
|
week = f["provenance"].get("evidence_week", f["last_seen"])
|
|
ev = load_evidence(week)
|
|
index = {row["id"]: row for group in ("commits", "ci", "issues", "pulls",
|
|
"releases", "declared")
|
|
for row in ev.get(group, [])}
|
|
index[f"w:{week}"] = ev.get("rollup", {})
|
|
print(json.dumps({"finding": f, "week": week,
|
|
"evidence": {eid: index.get(eid, "MISSING") for eid in f["evidence"]}},
|
|
indent=2, ensure_ascii=False))
|
|
return 0
|
|
|
|
if args.command == "draft":
|
|
return record_draft(args.week, args.file, args.author, args.model, note)
|
|
|
|
if args.command == "verdict":
|
|
return s.write_verdict(f"{args.finding}={args.verdict}", note, args.note, args.by)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|