Roll up fkst-hosted-default into develop

.fkst/workflows/cron-acceptance.toml

@@ -0,0 +1,69 @@
+description = "Milestone-24 acceptance: prove the scheduled-workflow machinery end to end."
+
+# The acceptance instance for #5846.
+#
+# It carries the SHAPE the issue specifies -- deterministic step, agentic step,
+# deterministic step that commits a cross-run ledger -- while deliberately
+# carrying NONE of the concrete instance. #5846 draws a content boundary: search
+# parameters, judgment criteria, destination-table identifiers, and
+# credential-broker service names are operator-supplied and must not enter this
+# repository. So the external API is replaced by locally generated candidates and
+# the destination table by a committed file.
+#
+# What that still proves is the bounded milestone: a due slot creating a run
+# issue, an idle session waking on it, `run` and `task` steps executing in
+# declared order, arguments substituted as data, a previously seeded cross-run
+# ledger suppressing repeats, at least one accepted row being committed, and one
+# fkst-cron-run:v1 record travelling back to release the schedule. The workflow
+# fails closed when the required prior-ledger suppression is not observable.
+#
+# What it does NOT prove is the operator's real workload -- live API pagination,
+# rate-limit backoff, and credential-broker delivery. Those need the concrete
+# instance, and this definition is the slot it drops into.
+
+[[step]]
+id = "collect"
+kind = "run"
+command = [
+  "python3",
+  ".fkst/workflows/cron-acceptance/collect.py",
+  "--topic",
+  "{{ topic }}",
+  "--count",
+  "{{ count }}",
+]
+timeout_secs = 300
+
+[[step]]
+id = "score"
+kind = "task"
+prompt = """
+Read `candidates.json` in the repository root. It is a JSON array of objects,
+each with an `id` and a `title`.
+
+Score every entry from 0 to 10 for how well its title matches this criterion:
+{{ criterion }}
+
+Write `scored.json` to the repository root: a JSON array, one object per input
+entry, each with exactly these keys:
+
+  id        the entry's id, unchanged
+  score     an integer 0-10
+  rationale one sentence, under 120 characters
+
+Emit the JSON array as the entire file content. Do not wrap it in a code fence
+and do not add commentary before or after it. Every input entry must appear
+exactly once in the output.
+"""
+timeout_secs = 900
+
+[[step]]
+id = "publish"
+kind = "run"
+command = [
+  "python3",
+  ".fkst/workflows/cron-acceptance/publish.py",
+  "--min-score",
+  "{{ min_score }}",
+]
+timeout_secs = 300

.fkst/workflows/cron-acceptance/collect.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Step 1 of the milestone-24 acceptance workflow: collect candidates.
+
+Deterministic. Emits `candidates.json` for the agentic scoring step, having
+first dropped every id the cross-run ledger already records as published.
+
+The ledger read is what makes a run after the seeded first run provably
+different, which is the property #5846 asks this step to demonstrate. The
+window slides by one id per run and overlaps the previous run by one. Acceptance
+therefore requires observable prior state: at least one id must be suppressed
+and at least one genuinely new id must be carried forward. A first run seeds the
+ledger but cannot, by itself, prove cross-run suppression.
+
+Candidates are generated locally rather than fetched. #5846 draws a content
+boundary around the concrete instance -- its API, search terms, and credentials
+are operator-supplied and must not enter this repository -- so this step proves
+the machinery around the fetch, and the operator swaps the fetch in.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import pathlib
+import re
+import sys
+
+import ledger
+
+LOG = logging.getLogger("collect")
+
+CANDIDATES = pathlib.Path("candidates.json")
+
+# One overlapping id per run: enough to prove the ledger suppressed something,
+# without suppressing so much that a run has nothing left to score.
+OVERLAP = 1
+
+# A slug keeps generated ids inside the character set the run record's `steps`
+# attribute can carry, and keeps a hostile `--topic` from reaching a filename.
+SLUG = re.compile(r"[^a-z0-9-]+")
+
+
+def slugify(topic: str) -> str:
+    """A filesystem- and marker-safe form of an operator-supplied topic."""
+    slug = SLUG.sub("-", topic.strip().lower()).strip("-")
+    if not slug:
+        raise ValueError(f"topic {topic!r} contains no usable characters")
+    return slug[:40]
+
+
+def build_window(slug: str, published: list[str], count: int) -> list[dict[str, str]]:
+    """The ids this run considers, overlapping the previous run by OVERLAP."""
+    start = max(0, len(published) - OVERLAP)
+    return [
+        {"id": f"{slug}-{index:04d}", "title": f"{slug} candidate {index}"}
+        for index in range(start, start + count)
+    ]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--topic", required=True, help="operator-supplied subject")
+    parser.add_argument("--count", required=True, help="window size for this run")
+    args = parser.parse_args()
+
+    try:
+        count = int(args.count)
+    except ValueError:
+        LOG.error("--count must be an integer, got %r", args.count)
+        return 2
+    if not 1 <= count <= 100:
+        LOG.error("--count must be between 1 and 100, got %d", count)
+        return 2
+
+    try:
+        slug = slugify(args.topic)
+    except ValueError as error:
+        LOG.error("%s", error)
+        return 2
+
+    try:
+        recorded, _ = ledger.read()
+    except ledger.LedgerError as error:
+        LOG.error("%s", error)
+        return 1
+    published = set(recorded)
+    window = build_window(slug, sorted(published), count)
+    fresh = [entry for entry in window if entry["id"] not in published]
+    suppressed = len(window) - len(fresh)
+
+    if suppressed < OVERLAP:
+        LOG.error(
+            "acceptance requires a seeded ledger that suppresses at least %d candidate(s); observed %d",
+            OVERLAP,
+            suppressed,
+        )
+        return 1
+    if not fresh:
+        LOG.error("acceptance requires at least one new candidate after ledger suppression")
+        return 1
+
+    CANDIDATES.write_text(json.dumps(fresh, indent=2) + "\n", encoding="utf-8")
+    LOG.info(
+        "topic=%s window=%d suppressed_by_ledger=%d carried_forward=%d -> %s",
+        slug,
+        len(window),
+        suppressed,
+        len(fresh),
+        CANDIDATES,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+    sys.exit(main())

.fkst/workflows/cron-acceptance/ledger.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""The cross-run ledger for the milestone-24 acceptance workflow.
+
+The ledger is the ONLY state that survives between runs, and #5846 requires it
+to be a committed file in the repository rather than a control-plane store --
+that is what keeps the scheduled-workflow capability stateless.
+
+It lives on its own branch, `cron-acceptance-ledger`, NOT on `develop`. Each run
+works in a fresh clone of the default branch, so a ledger committed to `develop`
+would need a pull request per run; `develop` is protected and the repository's
+own rules forbid pushing to it directly. A dedicated branch is writable, carries
+exactly two files, and keeps run bookkeeping out of the source history.
+
+Writes go through git plumbing (hash-object / mktree / commit-tree) rather than
+add+commit, so the run never switches the working tree it is executing from.
+
+The commit is IDEMPOTENT: if the resulting tree matches the branch tip's tree,
+nothing is committed. Re-running the same slot therefore cannot double-publish,
+which is the property #5846 asks for by name.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import subprocess
+
+LOG = logging.getLogger("ledger")
+
+BRANCH = "cron-acceptance-ledger"
+LEDGER_FILE = "ledger.json"
+PUBLISHED_FILE = "published.json"
+
+# Bound the ledger so a long-lived schedule cannot grow one file without limit.
+# The window only ever consults the tail, so older ids are safe to drop.
+MAX_IDS = 200
+
+
+class LedgerError(RuntimeError):
+    """A ledger operation failed in a way the run must not paper over."""
+
+
+def _git(*args: str, check: bool = True, stdin: str | None = None) -> str:
+    """Run one git command, returning stdout.
+
+    Never `shell=True`: arguments reach git as argv, so an operator-supplied
+    value cannot become shell syntax.
+    """
+    result = subprocess.run(
+        ["git", *args],
+        input=stdin,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if check and result.returncode != 0:
+        raise LedgerError(
+            f"git {' '.join(args)} failed ({result.returncode}): {result.stderr.strip()}"
+        )
+    return result.stdout
+
+
+def _tip() -> str | None:
+    """The ledger branch's current commit, or None when it does not exist yet."""
+    _git("fetch", "--quiet", "origin", BRANCH, check=False)
+    revision = _git("rev-parse", "--verify", "--quiet", "FETCH_HEAD", check=False).strip()
+    return revision or None
+
+
+def read() -> tuple[list[str], dict[str, dict]]:
+    """The published ids and the destination table, as of the branch tip.
+
+    A missing branch means the first run and yields empty state. A branch that
+    exists but whose contents will not parse is a HARD failure: treating
+    corruption as "nothing published yet" would re-publish the entire history,
+    the exact outcome the ledger exists to prevent.
+    """
+    tip = _tip()
+    if tip is None:
+        LOG.info("no %s branch yet; treating this as the first run", BRANCH)
+        return [], {}
+
+    ids = _read_json(tip, LEDGER_FILE, default=[])
+    if not isinstance(ids, list) or not all(isinstance(entry, str) for entry in ids):
+        raise LedgerError(f"{LEDGER_FILE} on {BRANCH} must be a JSON array of strings")
+
+    published = _read_json(tip, PUBLISHED_FILE, default={})
+    if not isinstance(published, dict):
+        raise LedgerError(f"{PUBLISHED_FILE} on {BRANCH} must be a JSON object keyed by id")
+
+    LOG.info("ledger at %s carries %d published id(s)", tip[:8], len(ids))
+    return ids, published
+
+
+def _read_json(tip: str, name: str, default):
+    """One file's decoded content from a commit, or `default` when absent."""
+    raw = _git("show", f"{tip}:{name}", check=False)
+    if not raw.strip():
+        return default
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError as error:
+        raise LedgerError(f"{name} on {BRANCH} is not valid JSON: {error}") from error
+
+
+def write(ids: list[str], published: dict[str, dict], message: str) -> bool:
+    """Commit the ledger and destination table. Returns False when unchanged."""
+    trimmed = ids[-MAX_IDS:]
+    entries = {
+        LEDGER_FILE: json.dumps(trimmed, indent=2) + "\n",
+        PUBLISHED_FILE: json.dumps(published, indent=2, sort_keys=True) + "\n",
+    }
+
+    lines = []
+    for name, content in sorted(entries.items()):
+        blob = _git("hash-object", "-w", "--stdin", stdin=content).strip()
+        lines.append(f"100644 blob {blob}\t{name}")
+    tree = _git("mktree", stdin="\n".join(lines) + "\n").strip()
+
+    tip = _tip()
+    if tip is not None and _git("rev-parse", f"{tip}^{{tree}}").strip() == tree:
+        LOG.info("ledger tree is unchanged; nothing to commit (idempotent re-run)")
+        return False
+
+    parents = ["-p", tip] if tip else []
+    commit = _git("commit-tree", tree, *parents, "-m", message).strip()
+    _git("push", "--quiet", "origin", f"{commit}:refs/heads/{BRANCH}")
+    LOG.info("committed ledger %s to %s (%d id(s) retained)", commit[:8], BRANCH, len(trimmed))
+    return True