coulomb/state-hub
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

CUST-WP-0061-T03: the promotion transition (statehub promote-intake)
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
Details
CI Smoke / container-smoke (push) Successful in 3s
Details

Browse source 
The mechanism named in canon/standards/work-record-types_v0.1.md:
"Promotion is a first-class transition... manual transcription of an
intake item into other kinds is a process defect." This is what AWQ-010
needed and didn't have -- a human/agent had to notice, transcribe, and
re-register it by hand. One call now does what that manual pass did.

scripts/promote_intake.py: intake.routed -> workplan | task | decision |
engagement.
  - workplan: new ADR-001 file at workplans/{ID}-{slug}.md, registered
    against the hub (repo+topic resolution, POST /workplans, frontmatter
    id write-back)
  - task: appended as a ```task``` block to an existing --workplan-file,
    registered via POST /tasks, reuses the existing
    _inject_task_id_into_block writeback helper
  - decision: appended as a ```yaml``` block with a fresh
    {PREFIX}-DEC-{YYYY}-{NNN} id to --target-file, registered the same
    way C-32 registers decisions (reuses _inject_yaml_block_field)
  - engagement: appended as a ```yaml``` block with a fresh
    {PREFIX}-ENG-{YYYY}-{NNN} id -- file-only, no hub entity exists yet
    (same honest deferral as C-32), reported not silently skipped

In every case the intake is closed with outcome=promoted and
promoted_to=<new canonical id>; the new record carries an
origin: "intake:<id>" back-link.

Wired as `statehub promote-intake <intake-id> --to <kind> --repo-slug
<slug> --repo-path <path> --domain <domain> [--target-file ...]
[--workplan-file ...]`, matching the CLI shape named in the workplan text.

17 tests: pure helpers (_slugify, _next_number, _append_yaml_block,
frontmatter injection) offline; full promote_intake() flow with the hub
API mocked.

Live-verified against the real running API/DB and a real repo
(binky-control), not just mocks -- and the live proof caught a real bug:
the first workplan-promotion run silently produced a false success (the
intake was closed outcome=promoted, but /workplans/ actually 422'd on a
missing repo_id that the code never resolved, so no workstream was ever
created). Fixed to resolve repo_id via /repos/{slug} and to raise loudly
on registration failure instead of writing a half-registered file
silently; locked in as two regression tests. Re-verified clean:
workplan promotion (CLI direct + through `statehub promote-intake`
itself) and decision promotion both proven end-to-end against the live
hub, with all scratch artefacts (files + hub rows) cleaned up afterward.
No regressions: full state-hub suite (271 tests across
test_promote_intake, test_intake, test_work_record_registration,
test_work_record_check, test_routers_core, test_consistency_check,
test_consistency_sweep, test_mcp_smoke, test_mcp_write_tools) green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-07-21 01:06:01 +02:00
parent aade470f4a
commit 3dbbc753bc
3 changed files with 694 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

40
custodian_cli.py
View file

@ -422,6 +422,30 @@ def cmd_fix_consistency(args: argparse.Namespace) -> None:
sys.exit(exit_code)
def cmd_promote_intake(args: argparse.Namespace) -> None:
"""Promote a routed intake into a workplan, task, decision, or engagement."""
script = STATE_HUB_DIR / "scripts" / "promote_intake.py"
if not script.exists():
print(f"ERROR: promote_intake.py not found at {script}")
sys.exit(1)
cmd = [
sys.executable, str(script), args.intake_id,
"--to", args.to,
"--repo-path", str(Path(args.repo_path).expanduser().resolve()),
"--repo-slug", args.repo_slug,
"--domain", args.domain,
"--api-base", args.api_base,
]
if args.target_file:
cmd.extend(["--target-file", args.target_file])
if args.workplan_file:
cmd.extend(["--workplan-file", args.workplan_file])
result = subprocess.run(cmd)
sys.exit(result.returncode)
def cmd_create_workstream(args: argparse.Namespace) -> None:
"""Create a workstream under a domain's topic."""
_api_get("/state/health")
@ -684,6 +708,20 @@ def main() -> None:
help="Preserve checker exit code 2 for warnings-only runs",
)
# promote-intake
promote = sub.add_parser(
"promote-intake",
help="Promote a routed intake into a workplan, task, decision, or engagement (CUST-WP-0061-T03)",
)
promote.add_argument("intake_id", help="UUID of the routed intake")
promote.add_argument("--to", required=True, choices=["workplan", "task", "decision", "engagement"])
promote.add_argument("--repo-path", default=os.getcwd(), help="Local checkout of the target repo")
promote.add_argument("--repo-slug", required=True, help="Registered repo slug")
promote.add_argument("--domain", required=True, help="Market domain slug, e.g. infotech")
promote.add_argument("--target-file", default=None, help="Repo-relative path (required for decision/engagement)")
promote.add_argument("--workplan-file", default=None, help="Repo-relative workplan file path (required for task)")
promote.add_argument("--api-base", default=API_BASE, help="State Hub API base URL")
# create-workstream
cws = sub.add_parser("create-workstream", help="Create a workstream under a domain topic")
cws.add_argument("--domain", required=True, help="Domain slug to create the workstream under")
@ -767,6 +805,8 @@ def main() -> None:
cmd_ingest_sbom(args)
elif args.command == "fix-consistency":
cmd_fix_consistency(args)
elif args.command == "promote-intake":
cmd_promote_intake(args)
elif args.command == "create-workstream":
cmd_create_workstream(args)
elif args.command == "create-task":

416
scripts/promote_intake.py Normal file
View file

@ -0,0 +1,416 @@
#!/usr/bin/env python3
"""Promote a routed intake work record into a workplan, task, decision, or
engagement — CUST-WP-0061-T03, the promotion transition named in
canon/standards/work-record-types_v0.1.md:
"Promotion is a first-class transition: intake.routed -> workplan |
task | decision | engagement, executed by one CLI/MCP call that writes
both artefacts, sets promoted_to/origin back-links, and re-syncs.
Manual transcription of an intake item into other kinds is a process
defect."
This is the mechanism AWQ-010 needed and didn't have: a human/agent had to
notice, transcribe, and re-register it by hand (binky-control, 2026-07-19).
One call does what that manual pass did.
Kinds:
workplan — new ADR-001 file at workplans/{ID}-{slug}.md, registered
against the hub the same way C-06 registers workplans.
task — appended as a ```task``` block to an existing workplan
file (--workplan-file), registered via POST /tasks.
decision — appended as a ```yaml``` block with a fresh {PREFIX}-DEC-
{YYYY}-{NNN} id to --target-file, registered the same way
C-32 registers decisions.
engagement — appended as a ```yaml``` block with a fresh {PREFIX}-ENG-
{YYYY}-{NNN} id to --target-file. File-only: no hub entity
exists for engagements yet (CUST-WP-0061 stage-3 seed) —
this is reported, not silently skipped.
In every case: the intake is closed with outcome=promoted and
promoted_to=<new canonical id>; the new record's file carries an
`origin: intake:<intake-id>` back-link.
"""
from __future__ import annotations
import argparse
import re
import sys
from datetime import date, datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from consistency_check import ( # noqa: E402
_api_get,
_api_patch,
_api_post,
_inject_task_id_into_block,
_inject_yaml_block_field,
infer_wp_prefix,
resolve_topic_domain_slug,
)
_SKIP_DIRS = frozenset({".git", "node_modules", ".venv", "history", "agents_backup"})
class PromotionError(Exception):
pass
def _slugify(title: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
return slug[:60].rstrip("-") or "untitled"
def _next_number(repo_dir: Path, id_regex: re.Pattern, *, width: int = 4) -> int:
"""Scan every .md file for ids matching id_regex (capture group 1 = the
numeric part); return the next free sequence number."""
max_n = 0
for md in repo_dir.rglob("*.md"):
if any(part in _SKIP_DIRS for part in md.parts):
continue
try:
text = md.read_text(errors="replace")
except OSError:
continue
for m in id_regex.finditer(text):
try:
max_n = max(max_n, int(m.group(1)))
except (ValueError, IndexError):
continue
return max_n + 1
def _fetch_intake(api_base: str, intake_id: str) -> dict:
intake = _api_get(api_base, f"/intakes/{intake_id}")
if not intake or (isinstance(intake, dict) and "_error" in intake):
raise PromotionError(f"could not fetch intake {intake_id}: {intake}")
if intake.get("status") != "routed":
raise PromotionError(
f"intake {intake_id} is '{intake.get('status')}', not 'routed' — "
f"route it first (POST /intakes/{intake_id}/route)"
)
return intake
def _close_intake(api_base: str, intake_id: str, canonical_id: str, to_kind: str) -> dict:
result = _api_post(api_base, f"/intakes/{intake_id}/close", {
"outcome": "promoted",
"promoted_to": canonical_id,
"note": f"promoted to {to_kind} {canonical_id}",
})
if not result or (isinstance(result, dict) and "_error" in result):
raise PromotionError(f"created {canonical_id} but failed to close intake: {result}")
return result
def _origin_line(intake_id: str) -> str:
return f'origin: "intake:{intake_id}"'
# ---------------------------------------------------------------------------
# workplan
# ---------------------------------------------------------------------------
def _promote_to_workplan(
api_base: str, repo_dir: Path, repo_slug: str, domain: str, intake: dict,
) -> tuple[str, Path]:
prefix = infer_wp_prefix(repo_dir, repo_slug)
id_re = re.compile(rf"\b{re.escape(prefix)}-(\d+)\b")
number = _next_number(repo_dir, id_re)
new_id = f"{prefix}-{str(number).zfill(4)}"
slug = _slugify(intake["title"])
workplans_dir = repo_dir / "workplans"
workplans_dir.mkdir(exist_ok=True)
new_file = workplans_dir / f"{new_id}-{slug}.md"
if new_file.exists():
raise PromotionError(f"target file already exists: {new_file}")
today = date.today().isoformat()
body = intake.get("description") or intake["title"]
content = (
f"---\n"
f'id: {new_id}\n'
f"type: workplan\n"
f'title: "{intake["title"]}"\n'
f"domain: {domain}\n"
f"repo: {repo_slug}\n"
f"status: proposed\n"
f"owner: codex\n"
f'created: "{today}"\n'
f'updated: "{today}"\n'
f"---\n\n"
f'# {intake["title"]}\n\n'
f"{body}\n\n"
f"**Promoted from intake** `{intake['id']}` — {_origin_line(intake['id'])}\n"
)
new_file.write_text(content, encoding="utf-8")
topics = _api_get(api_base, "/topics")
topic_domain = resolve_topic_domain_slug(domain, repo_market_domain=domain or None)
topic_id = None
if isinstance(topics, list):
for t in topics:
if t.get("domain_slug") == topic_domain:
topic_id = t["id"]
break
if topic_id is None:
raise PromotionError(
f"wrote {new_file.name} but no topic found for domain '{topic_domain}' "
f"— could not register the workstream; run fix-consistency to finish registration"
)
repo_record = _api_get(api_base, f"/repos/{repo_slug}")
if not repo_record or (isinstance(repo_record, dict) and "_error" in repo_record):
raise PromotionError(
f"wrote {new_file.name} but could not look up repo '{repo_slug}' "
f"({repo_record}) — could not register the workstream; "
f"run fix-consistency to finish registration"
)
ws = _api_post(api_base, "/workplans", {
"topic_id": topic_id,
"repo_id": repo_record["id"],
"title": intake["title"],
"slug": f"{repo_slug}-{slug}",
"status": "active",
"owner": "codex",
})
if not ws or (isinstance(ws, dict) and "_error" in ws):
raise PromotionError(
f"wrote {new_file.name} but workstream registration failed: {ws} "
f"— run fix-consistency to finish registration (C-06 will pick it up)"
)
_inject_field_in_frontmatter(new_file, "state_hub_workstream_id", ws["id"])
return new_id, new_file
def _inject_field_in_frontmatter(file_path: Path, field: str, value: str) -> None:
text = file_path.read_text(encoding="utf-8")
parts = text.split("---", 2)
if len(parts) < 3:
return
parts[1] = parts[1].rstrip() + f'\n{field}: "{value}"\n'
file_path.write_text("---".join(parts), encoding="utf-8")
# ---------------------------------------------------------------------------
# task
# ---------------------------------------------------------------------------
def _promote_to_task(api_base: str, workplan_file: Path, intake: dict) -> str:
if not workplan_file.is_file():
raise PromotionError(f"--workplan-file not found: {workplan_file}")
text = workplan_file.read_text(encoding="utf-8")
wp_id_match = re.search(r"^id:\s*(\S+)", text, re.MULTILINE)
if not wp_id_match:
raise PromotionError(f"could not find workplan id in {workplan_file}")
wp_id = wp_id_match.group(1).strip().strip('"')
ws_id_match = re.search(r'^state_hub_workstream_id:\s*"?([\w-]+)"?', text, re.MULTILINE)
if not ws_id_match:
raise PromotionError(
f"{workplan_file} has no state_hub_workstream_id — run fix-consistency on it first"
)
ws_id = ws_id_match.group(1)
existing_t_re = re.compile(rf"{re.escape(wp_id)}-T(\d+)")
number = _next_number(workplan_file.parent.parent, existing_t_re)
t_id = f"{wp_id}-T{str(number).zfill(2)}"
body = intake.get("description") or intake["title"]
section = (
f'\n## Task: {intake["title"]}\n\n'
f"{body}\n\n"
f"Promoted from intake `{intake['id']}`.\n\n"
f"```task\n"
f"id: {t_id}\n"
f"status: todo\n"
f"priority: medium\n"
f"```\n"
)
workplan_file.write_text(text.rstrip() + "\n" + section, encoding="utf-8")
task = _api_post(api_base, "/tasks", {
"workplan_id": ws_id,
"title": intake["title"],
"description": intake.get("description"),
"status": "todo",
"priority": "medium",
})
if task and "_error" not in task:
_inject_task_id_into_block(workplan_file, "state_hub_task_id", task["id"], t_id)
return t_id
# ---------------------------------------------------------------------------
# decision / engagement — share the "yaml block appended to a target file"
# shape; decision has a hub entity (C-32-style registration), engagement
# does not yet.
# ---------------------------------------------------------------------------
def _next_dec_or_eng_id(repo_dir: Path, prefix: str, kind_infix: str) -> str:
year = date.today().year
id_re = re.compile(rf"\b{re.escape(prefix)}-{kind_infix}-{year}-(\d+)\b")
number = _next_number(repo_dir, id_re, width=3)
return f"{prefix}-{kind_infix}-{year}-{str(number).zfill(3)}"
def _promote_to_decision(
api_base: str, repo_dir: Path, repo_slug: str, domain: str,
target_file: Path, intake: dict,
) -> str:
prefix = infer_wp_prefix(repo_dir, repo_slug).replace("-WP", "")
new_id = _next_dec_or_eng_id(repo_dir, prefix, "DEC")
_append_yaml_block(target_file, {
"id": new_id,
"title": intake["title"],
"status": "prepared",
"lane": intake.get("lane", "yellow"),
"agent_recommendation": intake.get("description") or "(fill in before founder review)",
"fallback_if_no_response": "no degradation — remains pending",
"created_at": date.today().isoformat(),
}, origin_intake_id=intake["id"])
topics = _api_get(api_base, "/topics")
topic_domain = resolve_topic_domain_slug(domain, repo_market_domain=domain or None)
topic_id = None
if isinstance(topics, list):
for t in topics:
if t.get("domain_slug") == topic_domain:
topic_id = t["id"]
break
if topic_id is None:
raise PromotionError(
f"wrote {new_id} to {target_file.name} but no topic found for domain "
f"'{topic_domain}' — run fix-consistency to finish registration"
)
decision = _api_post(api_base, "/decisions", {
"title": intake["title"],
"topic_id": topic_id,
"decision_type": "pending",
"description": intake.get("description"),
})
if decision and "_error" not in decision:
_inject_yaml_block_field(target_file, "state_hub_decision_id", decision["id"], new_id)
return new_id
def _promote_to_engagement(repo_dir: Path, repo_slug: str, target_file: Path, intake: dict) -> str:
prefix = infer_wp_prefix(repo_dir, repo_slug).replace("-WP", "")
new_id = _next_dec_or_eng_id(repo_dir, prefix, "ENG")
_append_yaml_block(target_file, {
"id": new_id,
"title": intake["title"],
"status": "queued",
"counterparty": "(fill in)",
"prepared_material": [],
"deadline_pressure": "none",
}, origin_intake_id=intake["id"])
return new_id
def _append_yaml_block(target_file: Path, fields: dict, *, origin_intake_id: str) -> None:
lines = ["```yaml"]
for k, v in fields.items():
if isinstance(v, str):
lines.append(f'{k}: "{v}"')
elif isinstance(v, list):
lines.append(f"{k}: {v!r}")
else:
lines.append(f"{k}: {v}")
lines.append(f'origin: "intake:{origin_intake_id}"')
lines.append("```")
block = "\n".join(lines) + "\n"
if target_file.exists():
existing = target_file.read_text(encoding="utf-8")
target_file.write_text(existing.rstrip() + "\n\n" + block, encoding="utf-8")
else:
target_file.parent.mkdir(parents=True, exist_ok=True)
target_file.write_text(block, encoding="utf-8")
# ---------------------------------------------------------------------------
# entry point
# ---------------------------------------------------------------------------
def promote_intake(
api_base: str,
intake_id: str,
to_kind: str,
repo_dir: Path,
repo_slug: str,
domain: str,
*,
target_file: str | None = None,
workplan_file: str | None = None,
) -> dict:
intake = _fetch_intake(api_base, intake_id)
if to_kind == "workplan":
canonical_id, new_file = _promote_to_workplan(api_base, repo_dir, repo_slug, domain, intake)
location = str(new_file.relative_to(repo_dir))
elif to_kind == "task":
if not workplan_file:
raise PromotionError("--to task requires --workplan-file")
wp_path = repo_dir / workplan_file
canonical_id = _promote_to_task(api_base, wp_path, intake)
location = workplan_file
elif to_kind == "decision":
if not target_file:
raise PromotionError("--to decision requires --target-file")
canonical_id = _promote_to_decision(
api_base, repo_dir, repo_slug, domain, repo_dir / target_file, intake
)
location = target_file
elif to_kind == "engagement":
if not target_file:
raise PromotionError("--to engagement requires --target-file")
canonical_id = _promote_to_engagement(repo_dir, repo_slug, repo_dir / target_file, intake)
location = target_file
else:
raise PromotionError(f"unsupported --to kind: {to_kind!r} (workplan|task|decision|engagement)")
_close_intake(api_base, intake_id, canonical_id, to_kind)
return {
"intake_id": intake_id,
"to_kind": to_kind,
"canonical_id": canonical_id,
"location": location,
}
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("intake_id")
ap.add_argument("--to", required=True, choices=["workplan", "task", "decision", "engagement"])
ap.add_argument("--repo-path", default=".", help="local checkout path of the target repo")
ap.add_argument("--repo-slug", required=True)
ap.add_argument("--domain", required=True, help="market domain slug, e.g. infotech")
ap.add_argument("--target-file", default=None, help="repo-relative path (decision/engagement)")
ap.add_argument("--workplan-file", default=None, help="repo-relative workplan file path (task)")
ap.add_argument("--api-base", default="http://127.0.0.1:8000")
args = ap.parse_args()
try:
result = promote_intake(
args.api_base, args.intake_id, args.to,
Path(args.repo_path).resolve(), args.repo_slug, args.domain,
target_file=args.target_file, workplan_file=args.workplan_file,
)
except PromotionError as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
print(f"Promoted intake {result['intake_id']} -> {result['to_kind']} {result['canonical_id']}")
print(f" file: {result['location']}")
print(" Review, commit, and push the change — promotion writes files but does not commit.")
if __name__ == "__main__":
main()

238
tests/test_promote_intake.py Normal file
View file

@ -0,0 +1,238 @@
"""Tests for the promotion transition (CUST-WP-0061-T03,
scripts/promote_intake.py): intake.routed -> workplan | task | decision |
engagement.
Pure file-manipulation logic (_slugify, _next_number, _append_yaml_block,
_inject_field_in_frontmatter) is tested offline. The full promote_intake()
flow is tested with the hub API layer mocked (monkeypatched _api_get/
_api_post) so these run without a live server; the real end-to-end proof
(actual DB writes) is documented in CUST-WP-0061's progress notes, mirroring
the live-verification discipline used for T01/T02.
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
import promote_intake as pi # noqa: E402
class TestSlugify:
def test_lowercases_and_hyphenates(self):
assert pi._slugify("Qonto MCP integration: bank account") == "qonto-mcp-integration-bank-account"
def test_truncates_long_titles(self):
long_title = "a " * 100
assert len(pi._slugify(long_title)) <= 60
def test_empty_title_gets_placeholder(self):
assert pi._slugify("!!!") == "untitled"
class TestNextNumber:
def test_finds_max_and_increments(self, tmp_path):
(tmp_path / "a.md").write_text("id: CUST-WP-0005\nid: CUST-WP-0012\n")
(tmp_path / "b.md").write_text("id: CUST-WP-0003\n")
import re
n = pi._next_number(tmp_path, re.compile(r"CUST-WP-(\d+)"))
assert n == 13
def test_empty_repo_starts_at_one(self, tmp_path):
import re
n = pi._next_number(tmp_path, re.compile(r"CUST-WP-(\d+)"))
assert n == 1
def test_skips_skip_dirs(self, tmp_path):
import re
skipped = tmp_path / ".git"
skipped.mkdir()
(skipped / "a.md").write_text("id: CUST-WP-9999\n")
n = pi._next_number(tmp_path, re.compile(r"CUST-WP-(\d+)"))
assert n == 1
class TestAppendYamlBlock:
def test_creates_new_file(self, tmp_path):
target = tmp_path / "DecisionQueue.md"
pi._append_yaml_block(target, {"id": "X-DEC-2026-001", "title": "test"}, origin_intake_id="abc-123")
text = target.read_text()
assert "id: \"X-DEC-2026-001\"" in text
assert 'origin: "intake:abc-123"' in text
def test_appends_to_existing_file(self, tmp_path):
target = tmp_path / "DecisionQueue.md"
target.write_text("# Decision Queue\n\nSome existing content.\n")
pi._append_yaml_block(target, {"id": "X-DEC-2026-002", "title": "second"}, origin_intake_id="def-456")
text = target.read_text()
assert "Some existing content." in text
assert "X-DEC-2026-002" in text
class TestInjectFieldInFrontmatter:
def test_injects_field_into_frontmatter(self, tmp_path):
f = tmp_path / "wp.md"
f.write_text("---\nid: CUST-WP-0099\ntitle: x\n---\n\nBody.\n")
pi._inject_field_in_frontmatter(f, "state_hub_workstream_id", "019f0000-0000-7000-8000-000000000000")
text = f.read_text()
assert 'state_hub_workstream_id: "019f0000-0000-7000-8000-000000000000"' in text
assert "Body." in text
class TestPromoteIntakeMocked:
"""Full promote_intake() flow with the hub API mocked."""
def _fake_routed_intake(self, **overrides):
base = {
"id": "019f0000-0000-7000-8000-000000000001",
"title": "Qonto MCP mailing — actionable",
"status": "routed",
"lane": "green",
"description": "found in mail triage",
}
base.update(overrides)
return base
def test_rejects_intake_not_routed(self, tmp_path, monkeypatch):
monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake(status="open"))
with pytest.raises(pi.PromotionError, match="not 'routed'"):
pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech")
def test_promote_to_workplan_writes_file_and_closes_intake(self, tmp_path, monkeypatch):
calls = []
def fake_get(api_base, path, *a, **k):
if path.startswith("/intakes/"):
return self._fake_routed_intake()
if path == "/topics":
return [{"id": "019ftopic", "domain_slug": "infotech"}]
if path == "/repos/testrepo":
return {"id": "019frepo", "slug": "testrepo"}
return None
def fake_post(api_base, path, body):
calls.append((path, body))
if path == "/workplans":
return {"id": "019fworkplan", "title": body["title"]}
if path.startswith("/intakes/") and path.endswith("/close"):
return {"id": "019f0000", "status": "closed", "outcome": "promoted"}
return {"id": "019fgeneric"}
monkeypatch.setattr(pi, "_api_get", fake_get)
monkeypatch.setattr(pi, "_api_post", fake_post)
result = pi.promote_intake(
"http://x", "019f0000-0000-7000-8000-000000000001", "workplan",
tmp_path, "testrepo", "infotech",
)
assert result["to_kind"] == "workplan"
new_file = tmp_path / result["location"]
assert new_file.is_file()
text = new_file.read_text()
assert "Qonto MCP mailing" in text
assert 'state_hub_workstream_id: "019fworkplan"' in text
assert "intake:019f0000-0000-7000-8000-000000000001" in text
close_calls = [c for c in calls if c[0].endswith("/close")]
assert len(close_calls) == 1
assert close_calls[0][1]["outcome"] == "promoted"
assert close_calls[0][1]["promoted_to"] == result["canonical_id"]
def test_promote_to_workplan_raises_loudly_when_repo_lookup_fails(self, tmp_path, monkeypatch):
"""Regression: the live proof against binky-control (2026-07-21) found
/workplans/ requires repo_id, which promote_to_workplan originally
never resolved -- the POST 422'd and the failure was silently
swallowed (file written, no state_hub_workstream_id, no error).
Must raise, not write a half-registered file silently."""
def fake_get(api_base, path, *a, **k):
if path.startswith("/intakes/"):
return self._fake_routed_intake()
if path == "/topics":
return [{"id": "019ftopic", "domain_slug": "infotech"}]
if path == "/repos/testrepo":
return {"_error": "404: not found"}
return None
monkeypatch.setattr(pi, "_api_get", fake_get)
monkeypatch.setattr(pi, "_api_post", lambda *a, **k: {"id": "x"})
with pytest.raises(pi.PromotionError, match="could not look up repo"):
pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech")
def test_promote_to_workplan_raises_loudly_when_workstream_post_fails(self, tmp_path, monkeypatch):
def fake_get(api_base, path, *a, **k):
if path.startswith("/intakes/"):
return self._fake_routed_intake()
if path == "/topics":
return [{"id": "019ftopic", "domain_slug": "infotech"}]
if path == "/repos/testrepo":
return {"id": "019frepo"}
return None
def fake_post(api_base, path, body):
if path == "/workplans":
return {"_error": "422: repo_id required"}
return {"id": "x"}
monkeypatch.setattr(pi, "_api_get", fake_get)
monkeypatch.setattr(pi, "_api_post", fake_post)
with pytest.raises(pi.PromotionError, match="workstream registration failed"):
pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech")
def test_promote_to_workplan_refuses_existing_file(self, tmp_path, monkeypatch):
monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake())
wp_dir = tmp_path / "workplans"
wp_dir.mkdir()
# pre-create the file the promotion would try to write
prefix = pi.infer_wp_prefix(tmp_path, "testrepo")
(wp_dir / f"{prefix}-0001-qonto-mcp-mailing-actionable.md").write_text("already here")
def fake_post(api_base, path, body):
return {"id": "x"}
monkeypatch.setattr(pi, "_api_post", fake_post)
with pytest.raises(pi.PromotionError, match="already exists"):
pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech")
def test_promote_to_decision_requires_target_file(self, tmp_path, monkeypatch):
monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake())
with pytest.raises(pi.PromotionError, match="--target-file"):
pi.promote_intake("http://x", "019f0000", "decision", tmp_path, "testrepo", "infotech")
def test_promote_to_task_requires_workplan_file(self, tmp_path, monkeypatch):
monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake())
with pytest.raises(pi.PromotionError, match="--workplan-file"):
pi.promote_intake("http://x", "019f0000", "task", tmp_path, "testrepo", "infotech")
def test_promote_to_engagement_is_file_only_no_hub_call(self, tmp_path, monkeypatch):
post_calls = []
def fake_get(api_base, path, *a, **k):
return self._fake_routed_intake()
def fake_post(api_base, path, body):
post_calls.append(path)
if path.startswith("/intakes/") and path.endswith("/close"):
return {"id": "x", "status": "closed"}
return {"id": "x"}
monkeypatch.setattr(pi, "_api_get", fake_get)
monkeypatch.setattr(pi, "_api_post", fake_post)
result = pi.promote_intake(
"http://x", "019f0000", "engagement", tmp_path, "testrepo", "infotech",
target_file="OfficeHourQueue.md",
)
assert result["to_kind"] == "engagement"
target = tmp_path / "OfficeHourQueue.md"
assert target.is_file()
assert "queued" in target.read_text()
# only the intake-close call should have hit the API — no
# engagement-creation endpoint exists yet
non_close_calls = [c for c in post_calls if not c.endswith("/close")]
assert non_close_calls == []