Verify offline scaffolds and preserve Forge qualifier receiving

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091

api/services/forge_projection.py

@@ -19,6 +19,7 @@ import subprocess
 import tempfile
 import uuid
 from dataclasses import dataclass, field
+from datetime import date
 from pathlib import Path
 from typing import Any
 
@@ -27,6 +28,10 @@ import yaml
 # Same derivation as ADR-007 / repo-manager, so a forge-derived projection and a
 # preliminary overlay compute identical identities for the same record.
 from api.services.task_record_id_backfill import qualify_task_id
+from api.work_record_flavor import (
+    FLAVOR_PROMOTION_REASONS, WORK_RECORD_FLAVORS,
+    normalize_flavor, normalize_promotion_reason,
+)
 
 _WORK_RECORD_NAMESPACE = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")
 _TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL)
@@ -129,6 +134,10 @@ class DerivedTask:
     needs_human: bool = False
     intervention_note: str | None = None
     blocking_reason: str | None = None
+    flavor: str | None = None
+    decision_id: str | None = None
+    blocked_on: str | None = None
+    wait_until: date | None = None
 
 
 @dataclass
@@ -142,6 +151,9 @@ class DerivedWorkplan:
     owner: str | None = None
     description: str | None = None
     tasks: list[DerivedTask] = field(default_factory=list)
+    flavor: str | None = None
+    flavor_promotion_reason: str | None = None
+    flavor_promoted_from: str | None = None
 
 
 @dataclass
@@ -188,6 +200,9 @@ class DerivedProjection:
                     "status": w.status,
                     "owner": w.owner,
                     "description": w.description,
+                    "flavor": w.flavor,
+                    "flavor_promotion_reason": w.flavor_promotion_reason,
+                    "flavor_promoted_from": w.flavor_promoted_from,
                     "relative_path": w.relative_path,
                     "archived": w.archived,
                     "tasks": [
@@ -201,6 +216,10 @@ class DerivedProjection:
                             "needs_human": t.needs_human,
                             "intervention_note": t.intervention_note,
                             "blocking_reason": t.blocking_reason,
+                            "flavor": t.flavor,
+                            "decision_id": t.decision_id,
+                            "blocked_on": t.blocked_on,
+                            "wait_until": t.wait_until.isoformat() if t.wait_until else None,
                         }
                         for t in w.tasks
                     ],
@@ -282,6 +301,56 @@ def _clean_task_description(raw: str) -> str | None:
     return "\n".join(lines).strip() or None
 
 
+
+def _source_flavor(value: Any, field_name: str = "flavor") -> str | None:
+    flavor = normalize_flavor(value)
+    if flavor is not None and flavor not in WORK_RECORD_FLAVORS:
+        raise ForgeDeriveError(f"Unknown source {field_name}; no projection was applied")
+    return flavor
+
+
+def _source_promotion_reason(value: Any) -> str | None:
+    reason = normalize_promotion_reason(value)
+    if reason is not None and reason not in FLAVOR_PROMOTION_REASONS:
+        raise ForgeDeriveError("Unknown source flavor_promotion_reason; no projection was applied")
+    return reason
+
+
+def _source_wait_until(value: Any) -> date | None:
+    if value is None or value == "":
+        return None
+    if type(value) is date:
+        return value
+    text = str(value).strip()
+    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
+        try:
+            return date.fromisoformat(text)
+        except ValueError:
+            pass
+    raise ForgeDeriveError("Source wait_until must be a valid YYYY-MM-DD date; no projection was applied")
+
+
+def _source_wait_text(value: Any) -> str | None:
+    if value is None:
+        return None
+    if not isinstance(value, str):
+        raise ForgeDeriveError("Source wait qualifier must be text; no projection was applied")
+    return value.strip() or None
+
+
+def _source_needs_human(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return False
+    text = str(value).strip().lower()
+    if text in {"true", "yes", "1", "on"}:
+        return True
+    if text in {"false", "no", "0", "off", ""}:
+        return False
+    raise ForgeDeriveError("Source needs_human must be boolean; no projection was applied")
+
+
 def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
     headings = [
         (m.start(), len(m.group(1)), m.group(2).strip()) for m in _HEADING_RE.finditer(body)
@@ -291,8 +360,10 @@ def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
     for index, m in enumerate(task_matches):
         try:
             block = yaml.safe_load(m.group(1).strip()) or {}
-        except yaml.YAMLError:
-            continue
+        except (yaml.YAMLError, ValueError) as exc:
+            # PyYAML raises ValueError for impossible bare ISO dates.
+            # Refuse the entire derive instead of dropping a stored gate.
+            raise ForgeDeriveError("Malformed source task metadata; no projection was applied") from exc
         if not isinstance(block, dict):
             continue
         rid = str(block.get("id") or "").strip()
@@ -333,17 +404,21 @@ def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
                 status=(str(block["status"]).strip() if block.get("status") else None),
                 priority=(str(block["priority"]).strip() if block.get("priority") else None),
                 description=description,
-                needs_human=bool(block.get("needs_human", False)),
+                needs_human=_source_needs_human(block.get("needs_human", False)),
                 intervention_note=(
                     str(block["intervention_note"])
                     if block.get("intervention_note") is not None
-                    else None
+                    else (str(block["blocking_reason"]) if _source_needs_human(block.get("needs_human")) and block.get("blocking_reason") is not None else None)
                 ),
                 blocking_reason=(
                     str(block["blocking_reason"])
                     if block.get("blocking_reason") is not None
                     else None
                 ),
+                flavor=_source_flavor(block.get("flavor")),
+                decision_id=_source_wait_text(block.get("decision_id")),
+                blocked_on=_source_wait_text(block.get("blocked_on")),
+                wait_until=_source_wait_until(block.get("wait_until")),
             )
         )
     return out
@@ -397,6 +472,9 @@ def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> Derive
                     else _workplan_description(body)
                 ),
                 tasks=_parse_tasks(body, rid),
+                flavor=_source_flavor(meta.get("flavor")),
+                flavor_promotion_reason=_source_promotion_reason(meta.get("flavor_promotion_reason")),
+                flavor_promoted_from=_source_flavor(meta.get("flavor_promoted_from"), "flavor_promoted_from"),
             )
         )
     proj.workplans.sort(key=lambda w: w.record_id)
@@ -611,6 +689,10 @@ def diff_against_hub(
         changed = {}
         if (cur.get("status") or None) != (w.status or None):
             changed["status"] = {"hub": cur.get("status"), "forge": w.status}
+        for field_name in ("flavor", "flavor_promotion_reason", "flavor_promoted_from"):
+            expected = getattr(w, field_name)
+            if cur.get(field_name) != expected:
+                changed[field_name] = {"hub": cur.get(field_name), "forge": expected}
         cur_path = cur.get("backing_relative_path") or None
         if cur_path != w.relative_path:
             changed["backing_relative_path"] = {"hub": cur_path, "forge": w.relative_path}
@@ -674,11 +756,25 @@ def diff_against_hub(
                 )
         for key, t in want.items():
             cur = have.get(key)
-            if cur and (cur.get("status") or None) != (t.status or None):
+            if not cur:
+                continue
+            changed = {}
+            if (cur.get("status") or None) != (t.status or None):
+                changed["status"] = {"hub": cur.get("status"), "forge": t.status}
+            for field_name in ("needs_human", "intervention_note", "blocking_reason",
+                               "flavor", "decision_id", "blocked_on", "wait_until"):
+                expected = getattr(t, field_name)
+                actual = cur.get(field_name, False if field_name == "needs_human" else None)
+                if isinstance(actual, date):
+                    actual = actual.isoformat()
+                if isinstance(expected, date):
+                    expected = expected.isoformat()
+                if actual != expected:
+                    changed[field_name] = {"hub": actual, "forge": expected}
+            if changed:
                 d.differing.append(
                     {"kind": "task", "record_id": t.record_id, "uuid": t.uuid,
-                     "workplan": w.record_id,
-                     "changed": {"status": {"hub": cur.get("status"), "forge": t.status}}}
+                     "workplan": w.record_id, "changed": changed}
                 )
     return d
 
@@ -822,6 +918,10 @@ def _sync_existing_workplan_tasks(
                 "needs_human": dt.needs_human,
                 "intervention_note": dt.intervention_note,
                 "blocking_reason": dt.blocking_reason,
+                "flavor": dt.flavor,
+                "decision_id": dt.decision_id,
+                "blocked_on": dt.blocked_on,
+                "wait_until": dt.wait_until,
             }
             st = _coerce_task_status(dt.status)
             if st is not None:
@@ -857,6 +957,11 @@ def _sync_existing_workplan_tasks(
         if getattr(ht, "blocking_reason", None) != dt.blocking_reason:
             ht.blocking_reason = dt.blocking_reason
             changed = True
+        for field_name in ("flavor", "decision_id", "blocked_on", "wait_until"):
+            expected = getattr(dt, field_name)
+            if getattr(ht, field_name, None) != expected:
+                setattr(ht, field_name, expected)
+                changed = True
         if dt.priority:
             try:
                 task_priority = TaskPriority(dt.priority.strip().lower())
@@ -1324,6 +1429,9 @@ async def reset_repository_projection(
                 description=w.description,
                 status=w.status or "proposed",
                 owner=w.owner,
+                flavor=w.flavor,
+                flavor_promotion_reason=w.flavor_promotion_reason,
+                flavor_promoted_from=w.flavor_promoted_from,
                 backing_filename=w.relative_path.rsplit("/", 1)[-1],
                 backing_relative_path=w.relative_path,
                 backing_archived=w.archived,
@@ -1346,6 +1454,10 @@ async def reset_repository_projection(
                         needs_human=t.needs_human,
                         intervention_note=t.intervention_note,
                         blocking_reason=t.blocking_reason,
+                        flavor=t.flavor,
+                        decision_id=t.decision_id,
+                        blocked_on=t.blocked_on,
+                        wait_until=t.wait_until,
                     )
                 )
             outcome.created.append(w.record_id)
@@ -1372,6 +1484,11 @@ async def reset_repository_projection(
         if w.status and row.status != w.status:
             row.status = w.status
             changed = True
+        for field_name in ("flavor", "flavor_promotion_reason", "flavor_promoted_from"):
+            expected = getattr(w, field_name)
+            if getattr(row, field_name, None) != expected:
+                setattr(row, field_name, expected)
+                changed = True
         if row.backing_relative_path != w.relative_path:
             row.backing_relative_path = w.relative_path
             row.backing_filename = w.relative_path.rsplit("/", 1)[-1]