CUST-WP-0055 T06: workplan-first curate decision scope

Dual-write workplan_id and legacy workstream_id on State Hub decisions;
add --workplan-id CLI flag. Archive grandfather notes (T07).

session_memory/curate/__main__.py

@@ -1,7 +1,7 @@
 """Curate entrypoint (T06): review detect candidates into the Pattern Catalog.
 
     python -m session_memory.curate [--config PATH] [--auto-approve] [--json]
-                                    [--workstream-id ID]
+                                    [--workplan-id ID] [--workstream-id ID]
 
 Refreshes candidate patterns (runs the detect pipeline), then drives them through
 the review workflow — **interactive** by default, or **batch** with
@@ -31,8 +31,8 @@ def _curate_paths(config: dict):
     catalog_dir = _expand(c.get("catalog_dir", "session_memory/catalog"))
     review_log = _expand(c.get("review_log", "session_memory/.store/reviews.jsonl"))
     queue = _expand(c.get("decision_queue", "session_memory/.store/decisions.queue.jsonl"))
-    ws_id = c.get("state_hub_workstream_id")
+    wp_id = c.get("state_hub_workplan_id") or c.get("state_hub_workstream_id")
-    return catalog_dir, review_log, queue, ws_id
+    return catalog_dir, review_log, queue, wp_id
 
 
 def _render_candidate(cand: dict, gate, existing) -> str:
@@ -96,7 +96,12 @@ def main(argv=None) -> int:
     ap.add_argument("--auto-approve", action="store_true",
                     help="batch mode: promote everything clearing the evidence bar")
     ap.add_argument("--min-frequency", type=int, default=2)
-    ap.add_argument("--workstream-id", default=None, help="hub workstream for decisions")
+    ap.add_argument("--workplan-id", default=None, help="hub workplan for decisions")
+    ap.add_argument(
+        "--workstream-id",
+        default=None,
+        help="legacy State Hub scope alias (workstream_id flag)",
+    )
     ap.add_argument("--json", action="store_true", help="emit machine-readable JSON")
     args = ap.parse_args(argv)
 
@@ -107,7 +112,8 @@ def main(argv=None) -> int:
     gate = gate_config(config)
     catalog = Catalog(catalog_dir)
     log = ReviewLog(review_log_path)
-    recorder = DecisionRecorder(queue_path, workstream_id=args.workstream_id or ws_id)
+    scope_id = args.workplan_id or args.workstream_id or ws_id
+    recorder = DecisionRecorder(queue_path, workplan_id=scope_id)
 
     decide = _auto_decider(gate) if args.auto_approve else _interactive_decider(gate, catalog)
     result = review(candidates, decide, catalog, log, gate=gate, recorder=recorder)

session_memory/curate/decisions.py

@@ -28,15 +28,16 @@ def _now() -> str:
 
 
 def build_decision(candidate: dict, action: str, rationale: str,
-                   *, workstream_id: Optional[str] = None,
+                   *, workplan_id: Optional[str] = None,
+                   workstream_id: Optional[str] = None,
                    decided_by: str = "curator") -> dict:
     """Shape a curate decision as a State Hub ``record_decision`` payload."""
     key = candidate["key"]
     verb = "Promote" if action == "approve" else "Reject"
-    return {
+    scope_id = workplan_id or workstream_id
+    payload = {
         "title": f"{verb} pattern candidate {key}",
         "decision_type": "made",
-        "workstream_id": workstream_id,
         "rationale": rationale,
         "decided_by": decided_by,
         "description": json.dumps({
@@ -46,6 +47,10 @@ def build_decision(candidate: dict, action: str, rationale: str,
         }, sort_keys=True),
         "recorded_at": _now(),
     }
+    if scope_id:
+        payload["workplan_id"] = scope_id
+        payload["workstream_id"] = scope_id
+    return payload
 
 
 @dataclass
@@ -54,14 +59,21 @@ class DecisionRecorder:
 
     queue_path: str
     sink: Optional[Sink] = None
+    workplan_id: Optional[str] = None
     workstream_id: Optional[str] = None
     decided_by: str = "curator"
     _queued: int = field(default=0, init=False)
 
     def record(self, candidate: dict, action: str, rationale: str) -> bool:
         """Record one decision. Returns True if the sink accepted it, else queued."""
-        payload = build_decision(candidate, action, rationale,
+        payload = build_decision(
-                                 workstream_id=self.workstream_id, decided_by=self.decided_by)
+            candidate,
+            action,
+            rationale,
+            workplan_id=self.workplan_id,
+            workstream_id=self.workstream_id,
+            decided_by=self.decided_by,
+        )
         if self.sink is not None:
             try:
                 self.sink(payload)

tests/test_curate_decisions.py

@@ -16,9 +16,10 @@ def _candidate(key="success:clean_pass:outcome"):
 
 
 def test_build_decision_payload_shape():
-    d = build_decision(_candidate(), "approve", "looks solid", workstream_id="ws-1")
+    d = build_decision(_candidate(), "approve", "looks solid", workplan_id="wp-1")
     assert d["decision_type"] == "made"
-    assert d["workstream_id"] == "ws-1"
+    assert d["workplan_id"] == "wp-1"
+    assert d["workstream_id"] == "wp-1"
     assert "Promote" in d["title"]
     assert d["rationale"] == "looks solid"
     assert "success:clean_pass:outcome" in d["description"]
@@ -61,7 +62,7 @@ def test_review_records_each_final_decision(tmp_path):
     cat = Catalog(str(tmp_path / "catalog"))
     log = ReviewLog(str(tmp_path / "reviews.jsonl"))
     captured = []
-    rec = DecisionRecorder(str(tmp_path / "q.jsonl"), sink=captured.append, workstream_id="ws")
+    rec = DecisionRecorder(str(tmp_path / "q.jsonl"), sink=captured.append, workplan_id="wp")
     cands = [_candidate("success:clean_pass:outcome"), _candidate("problem:abandoned:outcome")]
     review(cands, lambda c: (APPROVE if "success" in c["key"] else REJECT, "r"), cat, log,
            recorder=rec)