From 3dbbc753bc721011e38bd62adba46aa106a0f58a Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 21 Jul 2026 01:06:01 +0200 Subject: [PATCH] CUST-WP-0061-T03: the promotion transition (statehub promote-intake) 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=; the new record carries an origin: "intake:" back-link. Wired as `statehub promote-intake --to --repo-slug --repo-path --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 --- custodian_cli.py | 40 ++++ scripts/promote_intake.py | 416 +++++++++++++++++++++++++++++++++++ tests/test_promote_intake.py | 238 ++++++++++++++++++++ 3 files changed, 694 insertions(+) create mode 100644 scripts/promote_intake.py create mode 100644 tests/test_promote_intake.py diff --git a/custodian_cli.py b/custodian_cli.py index f0638b9..be404bb 100644 --- a/custodian_cli.py +++ b/custodian_cli.py @@ -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": diff --git a/scripts/promote_intake.py b/scripts/promote_intake.py new file mode 100644 index 0000000..b691c8e --- /dev/null +++ b/scripts/promote_intake.py @@ -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=; the new record's file carries an +`origin: intake:` 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() diff --git a/tests/test_promote_intake.py b/tests/test_promote_intake.py new file mode 100644 index 0000000..59fe2f1 --- /dev/null +++ b/tests/test_promote_intake.py @@ -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 == []