From 25d3dfa2578b0038b8c589e96eb6ec0505973657 Mon Sep 17 00:00:00 2001
From: firstmate-crewmate <brokentao@gmail.com>
Date: Fri, 18 Sep 2026 00:17:53 +0000
Subject: [PATCH 1/5] feat(helm): add missing-card and deleted-card phases to
 the Python sync planner

The canary check before merging PR #30 found that plan_board() only
processed cards already in the desired set, so a board card whose task
left every backlog was never closed, and a captain-deleted card on a
still-live task was silently recreated instead of held. Both phases
now mirror bin/fm-helm-lib.sh's missing_entries and deleted_entries,
proven byte-for-byte against the production jq planner with fixture
data only (fixture-owner, boards 999/1000).

Known remaining gaps, documented in docs/1.architecture and
docs/2.api: the jq side's malformed-body-ignore and new-card-intake
wake branches, dispatch-marker cleanup on a closed card, and
in-flight/blocked-specific hold wording are not yet ported.
---
 docs/1.architecture/3.helm-sync-python.md |   4 +-
 docs/2.api/3.helm-sync-python.md          |  10 +-
 python/helm_sync/model.py                 |  15 +-
 python/helm_sync/planner.py               | 157 +++++++++++++++++++++
 python/helm_sync/tests/jq_parity.py       | 158 +++++++++++++++++++++-
 5 files changed, 338 insertions(+), 6 deletions(-)

diff --git a/docs/1.architecture/3.helm-sync-python.md b/docs/1.architecture/3.helm-sync-python.md
index 1a6d2f4d3d5..accaf8f9c77 100644
--- a/docs/1.architecture/3.helm-sync-python.md
+++ b/docs/1.architecture/3.helm-sync-python.md
@@ -36,4 +36,6 @@ See [the Python Helm API reference](../2.api/3.helm-sync-python.md) for the publ
 This slice does not read or write Helm state, call GitHub Projects, execute plan actions, or provide a CLI.
 Those capabilities belong to later migration steps and must keep the current shell implementation active until their parity is proven.
 
-The planner also does not yet reconcile board cards that are missing or captain-deleted: it only plans cards whose task id is present in both the desired set and the current board snapshot. The production jq planner's retention, tombstone, and moved-card handling (`bin/fm-helm-lib.sh`) is deliberately deferred to the follow-up state/executor migration slice, not ported here.
+The planner now also closes a board card whose task has left every backlog (unless already Done) and holds rather than silently recreates a card that a captain deleted while its task is still live, matching `bin/fm-helm-lib.sh`'s `missing_entries` and `deleted_entries` phases at fixture scale.
+It does not yet raise jq's separate "captain added a card with no backlog task, run intake" wake, clear a stale dispatch marker on a closed card, or distinguish in-flight/blocked hold wording from queued.
+Project retention and the moved-card ledger (`bin/fm-helm-project-map.sh`) remain deferred to the follow-up state/executor migration slice, same as before.
diff --git a/docs/2.api/3.helm-sync-python.md b/docs/2.api/3.helm-sync-python.md
index 8d405bc23ff..50da518406d 100644
--- a/docs/2.api/3.helm-sync-python.md
+++ b/docs/2.api/3.helm-sync-python.md
@@ -181,6 +181,10 @@ Describes guarded field updates that preserve an Issue's title and body.
 
 Describes closing a previously synchronized card that is no longer in the desired set.
 
+### `SkipRecreatingDeletedCard(task, tombstone="", note="")`
+
+Describes declining to recreate a card whose cached item has left the board, leaving any captain hold or silent tombstone to the deleted-card phase instead.
+
 ### `RecordDispatchRequest(task, item, option, fingerprint)`
 
 Describes recording a new captain dispatch request.
@@ -193,7 +197,7 @@ Describes removing an obsolete dispatch request.
 
 Describes writing a valid captain Priority edit into its owning backlog.
 
-### `HoldDeletedTask(task, home_path, reason)`
+### `HoldDeletedTask(task, item, home_path, reason, wakes=(), divergence_changes=())`
 
 Describes a captain hold created after deletion of a live task's card.
 
@@ -336,5 +340,7 @@ The three-way comparison preserves captain edits, applies backlog edits that do
 
 ## Notes
 
-The planner returns typed actions, including `CreateDraft`, `UpdateDraft`, `UpdateIssueFields`, and `NoChange`.
+The planner returns typed actions, including `CreateDraft`, `UpdateDraft`, `UpdateIssueFields`, `NoChange`, `CloseMissingCard`, `SkipRecreatingDeletedCard`, `HoldDeletedTask`, and `KeepDeletedTombstone`.
+It closes a board card whose task has no backlog record anywhere (unless already Done), and it holds rather than recreates a card whose task is still live but whose card a captain deleted.
+It does not yet raise the separate jq intake wake for a captain-created card with no matching task, nor clear a stale dispatch marker when it closes a missing card.
 The package has no state adapter, GraphQL client, executor, or command-line entry point in this migration slice.
diff --git a/python/helm_sync/model.py b/python/helm_sync/model.py
index 3aec6fbe972..2431fef9cf3 100644
--- a/python/helm_sync/model.py
+++ b/python/helm_sync/model.py
@@ -451,6 +451,15 @@ def __post_init__(self) -> None:
         object.__setattr__(self, "acknowledge", MappingProxyType(dict(self.acknowledge)))
 
 
+@dataclass(frozen=True)
+class SkipRecreatingDeletedCard:
+    """Describe declining to recreate a card whose cached item left the board."""
+
+    task: TaskId
+    tombstone: str = ""
+    note: str = ""
+
+
 @dataclass(frozen=True)
 class RecordDispatchRequest:
     """Describe persisting a new captain dispatch request."""
@@ -482,8 +491,11 @@ class HoldDeletedTask:
     """Describe placing a captain hold after deletion of a live card."""
 
     task: TaskId
-    home_path: Path
+    item: ItemId
+    home_path: Path | None
     reason: str
+    wakes: tuple[WakeRequest, ...] = ()
+    divergence_changes: tuple[DivergenceChange, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -523,6 +535,7 @@ class ForgetDivergence:
     | UpdateDraft
     | UpdateIssueFields
     | CloseMissingCard
+    | SkipRecreatingDeletedCard
     | RecordDispatchRequest
     | ClearDispatchRequest
     | WritePriorityToBacklog
diff --git a/python/helm_sync/planner.py b/python/helm_sync/planner.py
index 1ae2d5e795f..4a5e6290165 100644
--- a/python/helm_sync/planner.py
+++ b/python/helm_sync/planner.py
@@ -14,6 +14,7 @@
     BoardSnapshot,
     CardBaseline,
     CardSnapshot,
+    CloseMissingCard,
     CreateDraft,
     DesiredCard,
     DivergenceChange,
@@ -22,11 +23,14 @@
     FieldWrite,
     FieldName,
     Fingerprint,
+    HoldDeletedTask,
     ItemId,
+    KeepDeletedTombstone,
     NoChange,
     OptionId,
     PlanAction,
     Signature,
+    SkipRecreatingDeletedCard,
     SyncState,
     TaskId,
     UpdateDraft,
@@ -607,6 +611,95 @@ def _unsupported_repo_note(
     return "", ()
 
 
+def _missing_phase_actions(
+    snapshot: BoardSnapshot, desired_by_task: Mapping[TaskId, DesiredCard]
+) -> list[PlanAction]:
+    """Close previously synchronized cards whose task left every backlog.
+
+    Mirrors ``bin/fm-helm-lib.sh``'s ``missing_entries`` close branch only: a
+    card with no valid task-id body, or a valid but never-before-seen task id
+    (a captain-created card with no backlog task), is left untouched here,
+    matching the jq side's "ignore" and "wake: run intake" branches, which
+    this port does not yet implement.
+    """
+    done_option = _option(snapshot.fields, "Status", "Done")
+    status_field = _field_id(snapshot.fields, "Status")
+    actions: list[PlanAction] = []
+    for card in snapshot.cards:
+        task = _item_task(card)
+        if task is None or task in desired_by_task:
+            continue
+        if _current_option(card, "Status") == done_option:
+            continue
+        actions.append(
+            CloseMissingCard(
+                task=task,
+                item=parse_item_id(card.item),
+                field_write=FieldWrite(status_field, "Status", "Done", OptionId(done_option)),
+                expected=card,
+                acknowledge={
+                    "new": False,
+                    "text": False,
+                    "fields": [{"name": "Status", "value": "Done", "option": done_option}],
+                },
+            )
+        )
+    return actions
+
+
+_DELETED_CARD_CHOICES = "the task is queued: cancel it (Done), mark it done, or was the card deleted by mistake"
+
+
+def _deleted_phase_actions(
+    snapshot: BoardSnapshot,
+    desired_by_task: Mapping[TaskId, DesiredCard],
+    state: SyncState,
+) -> list[PlanAction]:
+    """Preserve a captain-deleted card instead of silently recreating it.
+
+    Mirrors ``bin/fm-helm-lib.sh``'s ``deleted_entries``: a cached task whose
+    item id is gone from the board and whose task id has no replacement card
+    is either retained silently (task already Done) or raises a captain
+    hold. This port does not distinguish in-flight/blocked task states from
+    queued, since ``DesiredCard`` carries no such field; every hold uses the
+    same queued-style choices text.
+    """
+    item_set = {card.item for card in snapshot.cards}
+    line1_tasks = {task for card in snapshot.cards if (task := _item_task(card)) is not None}
+    actions: list[PlanAction] = []
+    for task, baseline in state.cards.items():
+        if baseline.board != snapshot.board:
+            continue
+        if baseline.item in item_set or task in line1_tasks:
+            continue
+        wanted = desired_by_task.get(task)
+        if wanted is None:
+            continue
+        if str(wanted.status) == "Done":
+            actions.append(KeepDeletedTombstone(task, baseline.item))
+            continue
+        fingerprint = str(baseline.item)
+        wakes: tuple[WakeRequest, ...] = ()
+        if not _divergence_exists(state.divergences, "card-deleted", task, baseline.item, fingerprint):
+            wakes = (
+                WakeRequest(
+                    f"helm-card-deleted:{task}",
+                    f"check: captain deleted Helm card {task} ({_DELETED_CARD_CHOICES})",
+                ),
+            )
+        actions.append(
+            HoldDeletedTask(
+                task=task,
+                item=baseline.item,
+                home_path=wanted.home_path,
+                reason=f"Helm card deleted; {_DELETED_CARD_CHOICES}.",
+                wakes=wakes,
+                divergence_changes=(DivergenceChange("card-deleted", "keep", str(baseline.item), fingerprint),),
+            )
+        )
+    return actions
+
+
 def plan_board(
     snapshot: BoardSnapshot,
     desired: Mapping[TaskId, DesiredCard] | Sequence[DesiredCard],
@@ -644,12 +737,22 @@ def plan_board(
         if task in cards_by_task:
             raise PlanError(f"duplicate Helm cards for {task}")
         cards_by_task[task] = card
+    item_set = {card.item for card in snapshot.cards}
 
     actions: list[PlanAction] = []
     for task, wanted in desired_by_task.items():
         current = cards_by_task.get(task)
         old = state.cards.get(task)
         if current is None:
+            tombstone = state.tombstones.get(task)
+            if tombstone is not None:
+                note, _ = _unsupported_repo_note(wanted, state)
+                actions.append(SkipRecreatingDeletedCard(task, f"{tombstone.task}\t{tombstone.item}", note))
+                continue
+            if old is not None and str(wanted.status) != "Done" and old.item and old.item not in item_set:
+                note, _ = _unsupported_repo_note(wanted, state)
+                actions.append(SkipRecreatingDeletedCard(task, "", note))
+                continue
             note, note_changes = _unsupported_repo_note(wanted, state)
             field_writes = _writes(
                 snapshot.fields,
@@ -698,6 +801,9 @@ def plan_board(
             )
         )
 
+    actions.extend(_missing_phase_actions(snapshot, desired_by_task))
+    actions.extend(_deleted_phase_actions(snapshot, desired_by_task, state))
+
     canonical = [
         {
             "item": str(card.item),
@@ -718,6 +824,14 @@ def _legacy_action_bytes(action: PlanAction) -> bytes:
     """Project a typed card decision to the live jq stream for parity tests."""
     if isinstance(action, CreateDraft):
         return _legacy_create_bytes(action)
+    if isinstance(action, CloseMissingCard):
+        return _legacy_missing_close_bytes(action)
+    if isinstance(action, KeepDeletedTombstone):
+        return _legacy_deleted_retain_bytes(action)
+    if isinstance(action, HoldDeletedTask):
+        return _legacy_deleted_hold_bytes(action)
+    if isinstance(action, SkipRecreatingDeletedCard):
+        return _legacy_skip_deleted_bytes(action)
     if not isinstance(action, (NoChange, UpdateDraft, UpdateIssueFields)):
         return b""
     baseline = action.baseline
@@ -828,6 +942,49 @@ def _legacy_create_bytes(action: CreateDraft) -> bytes:
     return "\0".join(values).encode("utf-8") + b"\0"
 
 
+def _legacy_skip_deleted_bytes(action: SkipRecreatingDeletedCard) -> bytes:
+    values = (
+        "record", "skip", str(action.task), "", "", "", "", "", "", "", "", "", "", "", "",
+        action.tombstone, "", "", "", "", "", action.note,
+    )
+    return "\0".join(values).encode("utf-8") + b"\0"
+
+
+def _legacy_missing_close_bytes(action: CloseMissingCard) -> bytes:
+    write = action.field_write
+    fields_text = _RS.join((write.field_id, write.name, write.value, str(write.option_id)))
+    values = (
+        "missing", "close", str(action.task), str(action.item), "", "", "", "",
+        fields_text, "", "", "", "", "", "", "", "",
+        _snapshot_json(action.expected), "", _json(dict(action.acknowledge)), "", "",
+    )
+    return "\0".join(values).encode("utf-8") + b"\0"
+
+
+def _legacy_deleted_retain_bytes(action: KeepDeletedTombstone) -> bytes:
+    tombstone = f"{action.task}\t{action.item}"
+    values = (
+        "deleted", "retain", str(action.task), "", "", "", "", "", "", "", "", "", "", "", "",
+        tombstone, "", "", "", "", "", "",
+    )
+    return "\0".join(values).encode("utf-8") + b"\0"
+
+
+def _legacy_deleted_hold_bytes(action: HoldDeletedTask) -> bytes:
+    tombstone = f"{action.task}\t{action.item}"
+    wakes_text = _US.join(_RS.join((wake.key, wake.payload)) for wake in action.wakes)
+    div_text = _US.join(
+        _RS.join((change.kind, change.action, change.item, change.fingerprint))
+        for change in action.divergence_changes
+    )
+    home = str(action.home_path) if action.home_path else ""
+    values = (
+        "deleted", "hold", str(action.task), str(action.item), "", "", "", "", "",
+        wakes_text, "", "", div_text, "", home, tombstone, action.reason, "", "", "", "", "",
+    )
+    return "\0".join(values).encode("utf-8") + b"\0"
+
+
 def _serialize_plan_for_jq_parity(plan: BoardPlan) -> bytes:
     """Encode a typed board plan in the existing jq format for parity tests."""
     return b"".join(_legacy_action_bytes(action) for action in plan.actions)
diff --git a/python/helm_sync/tests/jq_parity.py b/python/helm_sync/tests/jq_parity.py
index 924f51a500e..e161e4db862 100644
--- a/python/helm_sync/tests/jq_parity.py
+++ b/python/helm_sync/tests/jq_parity.py
@@ -24,6 +24,7 @@
     BoardSnapshot,
     CardBaseline,
     CardSnapshot,
+    DeletedCardTombstone,
     DesiredCard,
     DivergenceKey,
     DraftContent,
@@ -417,6 +418,9 @@ def _jq_plan_output(
     desired: DesiredCard,
     cards_tsv: str,
     divergences_text: str,
+    *,
+    state: str = "queued",
+    deleted_tsv: str = "",
 ) -> bytes:
     """Run the production planner with one synthetic board item."""
     with tempfile.TemporaryDirectory(prefix="helm-jq-plan-") as temp:
@@ -431,7 +435,7 @@ def _jq_plan_output(
         board_path.write_bytes(_compact_json(board))
         desired_record = {
             "id": str(desired.task),
-            "state": "queued",
+            "state": state,
             "home_path": "/fixture/main",
             "repo": desired.repository or "",
             "desired": {
@@ -448,7 +452,7 @@ def _jq_plan_output(
         }
         desired_path.write_bytes(_compact_json([desired_record]))
         cards_path.write_text(cards_tsv + "\n", encoding="utf-8")
-        deleted_path.write_text("", encoding="utf-8")
+        deleted_path.write_text(deleted_tsv, encoding="utf-8")
         markers_path.write_text("", encoding="utf-8")
         divergences_path.write_text(divergences_text, encoding="utf-8")
         body_hash = hashlib.sha256(desired.body.encode("utf-8")).hexdigest()
@@ -526,6 +530,82 @@ def _jq_plan_output(
         return result.stdout
 
 
+def _raw_field_nodes() -> list[dict[str, object]]:
+    """Build the raw board schema nodes matching the `_all_fields()` fixture schema."""
+    return [
+        {
+            "__typename": "ProjectV2SingleSelectField",
+            "id": field.id,
+            "name": field.name,
+            "options": [{"id": option_id, "name": option_name} for option_name, option_id in field.options.items()],
+        }
+        for field in _all_fields().values()
+    ]
+
+
+def _raw_draft_card(item: str, task: str, status: str, node: str) -> dict[str, object]:
+    """Build one raw board draft card claiming the given task in its body line 1."""
+    body = f"{TICK}{task}{TICK}\n\nbody"
+    return {
+        "id": item,
+        "content": {"__typename": "DraftIssue", "id": node, "title": "T", "body": body},
+        "fieldValues": {
+            "nodes": [
+                {"field": {"name": "Status"}, "name": status, "optionId": _status_id(status)},
+                {"field": {"name": "Project"}, "name": "sample", "optionId": "option_project_sample"},
+                {"field": {"name": "Kind"}, "name": "ship", "optionId": "option_kind_ship"},
+                {"field": {"name": "Priority"}, "name": "P3", "optionId": "option_priority_3"},
+            ]
+        },
+    }
+
+
+def _draft_card_snapshot(item: str, task: str, status: str, node: str) -> CardSnapshot:
+    """Build the Python-side counterpart of `_raw_draft_card`."""
+    body = f"{TICK}{task}{TICK}\n\nbody"
+    return CardSnapshot(
+        ItemId(item),
+        DraftContent(node, "T", body),
+        {
+            "Status": FieldValue("Status", status, OptionId(_status_id(status))),
+            "Project": FieldValue("Project", "sample", OptionId("option_project_sample")),
+            "Kind": FieldValue("Kind", "ship", OptionId("option_kind_ship")),
+            "Priority": FieldValue("Priority", "P3", OptionId("option_priority_3")),
+        },
+        TaskId(task),
+    )
+
+
+def _deleted_task_fixture(
+    task_name: str, item_name: str, status: str, *, state: str
+) -> tuple[DesiredCard, CardBaseline, BoardSnapshot, str]:
+    """Build a live task whose card was previously cached but is now gone from the board."""
+    task = TaskId(task_name)
+    item = ItemId(item_name)
+    board_ref = BoardRef(Owner("fixture-owner"), ProjectNumber(999))
+    body = f"{TICK}{task}{TICK}\n\nbody"
+    wanted = DesiredCard(
+        task,
+        HomeId("main"),
+        board_ref,
+        "T",
+        body,
+        StatusName(status),
+        KindName("ship"),
+        ProjectName("sample"),
+        PriorityName("P3"),
+        "3",
+        home_path=Path("/fixture/main"),
+    )
+    status_option = _status_id(status)
+    baseline = CardBaseline(task, item, board_ref, f"DRAFT_{item_name}", False, status_option, "option_priority_3", "T", body, "1699999999", True)
+    snapshot = BoardSnapshot(board_ref, (), _all_fields())
+    cards_tsv = "\t".join(
+        (str(task), str(item), f"DRAFT_{item_name}", "draft", status_option, "option_priority_3", _b64("T"), _b64(body), "1699999999", "fixture-owner", "999")
+    )
+    return wanted, baseline, snapshot, cards_tsv
+
+
 class HelmSyncPythonParityTests(unittest.TestCase):
     """Compare pure Python Helm modules with production jq behavior."""
 
@@ -685,6 +765,80 @@ def test_unsupported_repository_notice_on_new_card_matches_production_jq(self) -
         jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, divergence_text)
         self.assertEqual(jq_output, python_output)
 
+    def test_missing_card_with_no_backlog_record_closes_to_done(self) -> None:
+        """Close a previously synced card whose task left every backlog, matching missing_entries."""
+        snapshot, wanted, state, raw_board, cards_tsv, divergence_text = _planner_fixture()
+        missing_task, missing_item, missing_node = "fixture-missing-task", "PVTI_missing", "DRAFT_missing"
+        missing_card = _draft_card_snapshot(missing_item, missing_task, "Queued", missing_node)
+        extended_snapshot = BoardSnapshot(snapshot.board, snapshot.cards + (missing_card,), snapshot.fields)
+        raw_board["data"]["user"]["projectV2"]["items"]["nodes"].append(
+            _raw_draft_card(missing_item, missing_task, "Queued", missing_node)
+        )
+        missing_body = f"{TICK}{missing_task}{TICK}\n\nbody"
+        missing_cache_row = "\t".join(
+            (missing_task, missing_item, missing_node, "draft", "option_status_queued", "option_priority_3",
+             _b64("T"), _b64(missing_body), "1699999999", "fixture-owner", "999")
+        )
+        merged_cards_tsv = cards_tsv + "\n" + missing_cache_row
+        plan = plan_board(extended_snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        jq_output = _jq_plan_output(raw_board, wanted, merged_cards_tsv, divergence_text)
+        self.assertEqual(jq_output, python_output)
+
+    def test_missing_card_already_done_is_left_alone(self) -> None:
+        """Leave an already-Done orphaned card untouched, matching missing_entries' skip."""
+        snapshot, wanted, state, raw_board, cards_tsv, divergence_text = _planner_fixture()
+        missing_task, missing_item, missing_node = "fixture-missing-done-task", "PVTI_missing_done", "DRAFT_missing_done"
+        missing_card = _draft_card_snapshot(missing_item, missing_task, "Done", missing_node)
+        extended_snapshot = BoardSnapshot(snapshot.board, snapshot.cards + (missing_card,), snapshot.fields)
+        raw_board["data"]["user"]["projectV2"]["items"]["nodes"].append(
+            _raw_draft_card(missing_item, missing_task, "Done", missing_node)
+        )
+        plan = plan_board(extended_snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, divergence_text)
+        self.assertEqual(jq_output, python_output)
+
+    def test_deleted_card_for_a_live_task_raises_a_captain_hold(self) -> None:
+        """Preserve a captain's card deletion and raise one hold instead of recreating it."""
+        wanted, baseline, snapshot, cards_tsv = _deleted_task_fixture(
+            "fixture-deleted-task", "PVTI_deleted_old", "Queued", state="queued"
+        )
+        state = SyncState(cards={wanted.task: baseline})
+        plan = plan_board(snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        raw_board = {"data": {"user": {"projectV2": {"fields": {"nodes": _raw_field_nodes()}, "items": {"nodes": []}}}}}
+        jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, "", state="queued")
+        self.assertEqual(jq_output, python_output)
+
+    def test_deleted_card_for_an_already_done_task_is_retained_silently(self) -> None:
+        """Retain a tombstoned deletion for a Done task without recreating the card or waking."""
+        wanted, baseline, snapshot, cards_tsv = _deleted_task_fixture(
+            "fixture-deleted-done-task", "PVTI_deleted_done", "Done", state="done"
+        )
+        state = SyncState(
+            cards={wanted.task: baseline},
+            tombstones={wanted.task: DeletedCardTombstone(wanted.task, baseline.item)},
+        )
+        plan = plan_board(snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        raw_board = {"data": {"user": {"projectV2": {"fields": {"nodes": _raw_field_nodes()}, "items": {"nodes": []}}}}}
+        deleted_tsv = f"{wanted.task}\t{baseline.item}\n"
+        jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, "", state="done", deleted_tsv=deleted_tsv)
+        self.assertEqual(jq_output, python_output)
+
+    def test_deleted_card_for_a_done_task_without_a_tombstone_still_recreates(self) -> None:
+        """Match jq's existing recreate behavior for a Done task never previously tombstoned."""
+        wanted, baseline, snapshot, cards_tsv = _deleted_task_fixture(
+            "fixture-deleted-done-notomb-task", "PVTI_deleted_done_notomb", "Done", state="done"
+        )
+        state = SyncState(cards={wanted.task: baseline})
+        plan = plan_board(snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        raw_board = {"data": {"user": {"projectV2": {"fields": {"nodes": _raw_field_nodes()}, "items": {"nodes": []}}}}}
+        jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, "", state="done")
+        self.assertEqual(jq_output, python_output)
+
     def test_config_boundary_is_opt_in_and_validates_board_identity(self) -> None:
         """Return Disabled without inputs when absent and reject invalid config values."""
         with tempfile.TemporaryDirectory(prefix="helm-settings-") as temp:

From 27ce5597b88110d2c0b53944675801961115906c Mon Sep 17 00:00:00 2001
From: firstmate-crewmate <brokentao@gmail.com>
Date: Fri, 18 Sep 2026 00:33:35 +0000
Subject: [PATCH 2/5] no-mistakes(review): fix(helm): wake intake for unseen
 orphan cards and retain captain-held deletions

---
 python/helm_sync/desired.py         |  3 ++
 python/helm_sync/model.py           | 12 ++++++
 python/helm_sync/planner.py         | 66 +++++++++++++++++++++++------
 python/helm_sync/tests/jq_parity.py | 36 +++++++++++++++-
 4 files changed, 102 insertions(+), 15 deletions(-)

diff --git a/python/helm_sync/desired.py b/python/helm_sync/desired.py
index 1dc37d244fe..75f0735f28b 100644
--- a/python/helm_sync/desired.py
+++ b/python/helm_sync/desired.py
@@ -150,6 +150,8 @@ def render_card(
         home_path = None
     board = board_for_project(project_name, route_map, default_board)
     note = "" if project_name is not None else f"fm-helm-sync: unsupported repository {repo or ''} for {task}; using other"
+    raw_hold_kind = _value(record, "hold_kind")
+    hold_kind = str(raw_hold_kind) if raw_hold_kind else None
     return DesiredCard(
         task=task,
         home=HomeId(str(home_id)),
@@ -165,6 +167,7 @@ def render_card(
         home_path=home_path,
         note=note,
         repository=repo,
+        hold_kind=hold_kind,
     )
 
 
diff --git a/python/helm_sync/model.py b/python/helm_sync/model.py
index 2431fef9cf3..7d7bce1bbe4 100644
--- a/python/helm_sync/model.py
+++ b/python/helm_sync/model.py
@@ -181,6 +181,7 @@ class DesiredCard:
     home_path: Path | None = None
     note: str = ""
     repository: str | None = None
+    hold_kind: str | None = None
 
 
 @dataclass(frozen=True)
@@ -451,6 +452,16 @@ def __post_init__(self) -> None:
         object.__setattr__(self, "acknowledge", MappingProxyType(dict(self.acknowledge)))
 
 
+@dataclass(frozen=True)
+class WakeMissingCard:
+    """Describe leaving a never-before-seen orphan card untouched but waking intake."""
+
+    task: TaskId
+    item: ItemId
+    wakes: tuple[WakeRequest, ...] = ()
+    divergence_changes: tuple[DivergenceChange, ...] = ()
+
+
 @dataclass(frozen=True)
 class SkipRecreatingDeletedCard:
     """Describe declining to recreate a card whose cached item left the board."""
@@ -535,6 +546,7 @@ class ForgetDivergence:
     | UpdateDraft
     | UpdateIssueFields
     | CloseMissingCard
+    | WakeMissingCard
     | SkipRecreatingDeletedCard
     | RecordDispatchRequest
     | ClearDispatchRequest
diff --git a/python/helm_sync/planner.py b/python/helm_sync/planner.py
index 4a5e6290165..433ce2b4c1a 100644
--- a/python/helm_sync/planner.py
+++ b/python/helm_sync/planner.py
@@ -35,6 +35,7 @@
     TaskId,
     UpdateDraft,
     UpdateIssueFields,
+    WakeMissingCard,
     WakeRequest,
     parse_item_id,
     parse_task_id,
@@ -612,15 +613,15 @@ def _unsupported_repo_note(
 
 
 def _missing_phase_actions(
-    snapshot: BoardSnapshot, desired_by_task: Mapping[TaskId, DesiredCard]
+    snapshot: BoardSnapshot, desired_by_task: Mapping[TaskId, DesiredCard], state: SyncState
 ) -> list[PlanAction]:
-    """Close previously synchronized cards whose task left every backlog.
+    """Close or wake on previously seen and never-before-seen orphan cards.
 
-    Mirrors ``bin/fm-helm-lib.sh``'s ``missing_entries`` close branch only: a
-    card with no valid task-id body, or a valid but never-before-seen task id
-    (a captain-created card with no backlog task), is left untouched here,
-    matching the jq side's "ignore" and "wake: run intake" branches, which
-    this port does not yet implement.
+    Mirrors ``bin/fm-helm-lib.sh``'s ``missing_entries``: a card with no valid
+    task-id body is left untouched (matching jq's "ignore" branch, which this
+    port does not yet raise a note for), a valid but never-before-seen task id
+    (a captain-created card with no backlog task) wakes intake instead of
+    being closed, and every other orphaned, previously-cached task is closed.
     """
     done_option = _option(snapshot.fields, "Status", "Done")
     status_field = _field_id(snapshot.fields, "Status")
@@ -631,10 +632,30 @@ def _missing_phase_actions(
             continue
         if _current_option(card, "Status") == done_option:
             continue
+        item = parse_item_id(card.item)
+        if task not in state.cards:
+            fingerprint = str(item)
+            wakes: tuple[WakeRequest, ...] = ()
+            if not _divergence_exists(state.divergences, "new-card", task, item, fingerprint):
+                wakes = (
+                    WakeRequest(
+                        f"helm-new-card:{task}",
+                        f"check: captain added Helm card {task} with no backlog task; run intake",
+                    ),
+                )
+            actions.append(
+                WakeMissingCard(
+                    task=task,
+                    item=item,
+                    wakes=wakes,
+                    divergence_changes=(DivergenceChange("new-card", "keep", str(item), fingerprint),),
+                )
+            )
+            continue
         actions.append(
             CloseMissingCard(
                 task=task,
-                item=parse_item_id(card.item),
+                item=item,
                 field_write=FieldWrite(status_field, "Status", "Done", OptionId(done_option)),
                 expected=card,
                 acknowledge={
@@ -659,10 +680,11 @@ def _deleted_phase_actions(
 
     Mirrors ``bin/fm-helm-lib.sh``'s ``deleted_entries``: a cached task whose
     item id is gone from the board and whose task id has no replacement card
-    is either retained silently (task already Done) or raises a captain
-    hold. This port does not distinguish in-flight/blocked task states from
-    queued, since ``DesiredCard`` carries no such field; every hold uses the
-    same queued-style choices text.
+    is either retained silently (task already Done, or already under an
+    existing captain hold) or raises a captain hold. This port does not
+    distinguish in-flight/blocked task states from queued, since
+    ``DesiredCard`` carries no such field; every hold uses the same
+    queued-style choices text.
     """
     item_set = {card.item for card in snapshot.cards}
     line1_tasks = {task for card in snapshot.cards if (task := _item_task(card)) is not None}
@@ -675,7 +697,7 @@ def _deleted_phase_actions(
         wanted = desired_by_task.get(task)
         if wanted is None:
             continue
-        if str(wanted.status) == "Done":
+        if str(wanted.status) == "Done" or wanted.hold_kind == "captain":
             actions.append(KeepDeletedTombstone(task, baseline.item))
             continue
         fingerprint = str(baseline.item)
@@ -801,7 +823,7 @@ def plan_board(
             )
         )
 
-    actions.extend(_missing_phase_actions(snapshot, desired_by_task))
+    actions.extend(_missing_phase_actions(snapshot, desired_by_task, state))
     actions.extend(_deleted_phase_actions(snapshot, desired_by_task, state))
 
     canonical = [
@@ -826,6 +848,8 @@ def _legacy_action_bytes(action: PlanAction) -> bytes:
         return _legacy_create_bytes(action)
     if isinstance(action, CloseMissingCard):
         return _legacy_missing_close_bytes(action)
+    if isinstance(action, WakeMissingCard):
+        return _legacy_missing_wake_bytes(action)
     if isinstance(action, KeepDeletedTombstone):
         return _legacy_deleted_retain_bytes(action)
     if isinstance(action, HoldDeletedTask):
@@ -961,6 +985,20 @@ def _legacy_missing_close_bytes(action: CloseMissingCard) -> bytes:
     return "\0".join(values).encode("utf-8") + b"\0"
 
 
+def _legacy_missing_wake_bytes(action: WakeMissingCard) -> bytes:
+    wakes_text = _US.join(_RS.join((wake.key, wake.payload)) for wake in action.wakes)
+    div_text = _US.join(
+        _RS.join((change.kind, change.action, change.item, change.fingerprint))
+        for change in action.divergence_changes
+    )
+    values = (
+        "missing", "wake", str(action.task), str(action.item), "", "", "", "",
+        "", wakes_text, "", "", div_text, "", "", "", "",
+        "", "", "", "", "",
+    )
+    return "\0".join(values).encode("utf-8") + b"\0"
+
+
 def _legacy_deleted_retain_bytes(action: KeepDeletedTombstone) -> bytes:
     tombstone = f"{action.task}\t{action.item}"
     values = (
diff --git a/python/helm_sync/tests/jq_parity.py b/python/helm_sync/tests/jq_parity.py
index e161e4db862..1348015dbdf 100644
--- a/python/helm_sync/tests/jq_parity.py
+++ b/python/helm_sync/tests/jq_parity.py
@@ -421,6 +421,7 @@ def _jq_plan_output(
     *,
     state: str = "queued",
     deleted_tsv: str = "",
+    hold_kind: str = "",
 ) -> bytes:
     """Run the production planner with one synthetic board item."""
     with tempfile.TemporaryDirectory(prefix="helm-jq-plan-") as temp:
@@ -438,6 +439,7 @@ def _jq_plan_output(
             "state": state,
             "home_path": "/fixture/main",
             "repo": desired.repository or "",
+            "hold_kind": hold_kind,
             "desired": {
                 "title": desired.title,
                 "body": desired.body,
@@ -577,7 +579,7 @@ def _draft_card_snapshot(item: str, task: str, status: str, node: str) -> CardSn
 
 
 def _deleted_task_fixture(
-    task_name: str, item_name: str, status: str, *, state: str
+    task_name: str, item_name: str, status: str, *, state: str, hold_kind: str = ""
 ) -> tuple[DesiredCard, CardBaseline, BoardSnapshot, str]:
     """Build a live task whose card was previously cached but is now gone from the board."""
     task = TaskId(task_name)
@@ -596,6 +598,7 @@ def _deleted_task_fixture(
         PriorityName("P3"),
         "3",
         home_path=Path("/fixture/main"),
+        hold_kind=hold_kind or None,
     )
     status_option = _status_id(status)
     baseline = CardBaseline(task, item, board_ref, f"DRAFT_{item_name}", False, status_option, "option_priority_3", "T", body, "1699999999", True)
@@ -780,11 +783,30 @@ def test_missing_card_with_no_backlog_record_closes_to_done(self) -> None:
              _b64("T"), _b64(missing_body), "1699999999", "fixture-owner", "999")
         )
         merged_cards_tsv = cards_tsv + "\n" + missing_cache_row
+        missing_baseline = CardBaseline(
+            TaskId(missing_task), ItemId(missing_item), snapshot.board, missing_node, False,
+            "option_status_queued", "option_priority_3", "T", missing_body, "1699999999", True,
+        )
+        state = SyncState(cards={**state.cards, TaskId(missing_task): missing_baseline})
         plan = plan_board(extended_snapshot, (wanted,), state, force=True, epoch="1700000001")
         python_output = _serialize_plan_for_jq_parity(plan)
         jq_output = _jq_plan_output(raw_board, wanted, merged_cards_tsv, divergence_text)
         self.assertEqual(jq_output, python_output)
 
+    def test_missing_card_never_before_seen_wakes_intake_without_closing(self) -> None:
+        """Wake intake instead of closing an orphan card jq's old cache never saw."""
+        snapshot, wanted, state, raw_board, cards_tsv, divergence_text = _planner_fixture()
+        missing_task, missing_item, missing_node = "fixture-new-orphan-task", "PVTI_new_orphan", "DRAFT_new_orphan"
+        missing_card = _draft_card_snapshot(missing_item, missing_task, "Queued", missing_node)
+        extended_snapshot = BoardSnapshot(snapshot.board, snapshot.cards + (missing_card,), snapshot.fields)
+        raw_board["data"]["user"]["projectV2"]["items"]["nodes"].append(
+            _raw_draft_card(missing_item, missing_task, "Queued", missing_node)
+        )
+        plan = plan_board(extended_snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, divergence_text)
+        self.assertEqual(jq_output, python_output)
+
     def test_missing_card_already_done_is_left_alone(self) -> None:
         """Leave an already-Done orphaned card untouched, matching missing_entries' skip."""
         snapshot, wanted, state, raw_board, cards_tsv, divergence_text = _planner_fixture()
@@ -827,6 +849,18 @@ def test_deleted_card_for_an_already_done_task_is_retained_silently(self) -> Non
         jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, "", state="done", deleted_tsv=deleted_tsv)
         self.assertEqual(jq_output, python_output)
 
+    def test_deleted_card_under_existing_captain_hold_is_retained_silently(self) -> None:
+        """Retain a deletion silently under a live captain hold, even though the task isn't Done."""
+        wanted, baseline, snapshot, cards_tsv = _deleted_task_fixture(
+            "fixture-deleted-hold-task", "PVTI_deleted_hold", "Waiting on you", state="queued", hold_kind="captain"
+        )
+        state = SyncState(cards={wanted.task: baseline})
+        plan = plan_board(snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        raw_board = {"data": {"user": {"projectV2": {"fields": {"nodes": _raw_field_nodes()}, "items": {"nodes": []}}}}}
+        jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, "", state="queued", hold_kind="captain")
+        self.assertEqual(jq_output, python_output)
+
     def test_deleted_card_for_a_done_task_without_a_tombstone_still_recreates(self) -> None:
         """Match jq's existing recreate behavior for a Done task never previously tombstoned."""
         wanted, baseline, snapshot, cards_tsv = _deleted_task_fixture(

From 55e8487fadf710df58b3ad6a3a0b4bc6e3e556b2 Mon Sep 17 00:00:00 2001
From: firstmate-crewmate <brokentao@gmail.com>
Date: Fri, 18 Sep 2026 00:41:35 +0000
Subject: [PATCH 3/5] no-mistakes(review): fix(helm): close orphan cards on
 first-ever sync, not wake

---
 python/helm_sync/model.py           |  1 +
 python/helm_sync/planner.py         |  5 +++--
 python/helm_sync/tests/jq_parity.py | 18 +++++++++++++++++-
 3 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/python/helm_sync/model.py b/python/helm_sync/model.py
index 7d7bce1bbe4..b8f8e1905b7 100644
--- a/python/helm_sync/model.py
+++ b/python/helm_sync/model.py
@@ -308,6 +308,7 @@ class SyncState:
     poll_signatures: Mapping[BoardRef, Signature] = field(default_factory=dict)
     completed_input_hash: InputHash | None = None
     forced_resume: bool = False
+    cache_existed: bool = True
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "cards", MappingProxyType(dict(self.cards)))
diff --git a/python/helm_sync/planner.py b/python/helm_sync/planner.py
index 433ce2b4c1a..da6270079f6 100644
--- a/python/helm_sync/planner.py
+++ b/python/helm_sync/planner.py
@@ -621,7 +621,8 @@ def _missing_phase_actions(
     task-id body is left untouched (matching jq's "ignore" branch, which this
     port does not yet raise a note for), a valid but never-before-seen task id
     (a captain-created card with no backlog task) wakes intake instead of
-    being closed, and every other orphaned, previously-cached task is closed.
+    being closed when the identity cache already existed before this run,
+    and every other orphaned task is closed.
     """
     done_option = _option(snapshot.fields, "Status", "Done")
     status_field = _field_id(snapshot.fields, "Status")
@@ -633,7 +634,7 @@ def _missing_phase_actions(
         if _current_option(card, "Status") == done_option:
             continue
         item = parse_item_id(card.item)
-        if task not in state.cards:
+        if state.cache_existed and task not in state.cards:
             fingerprint = str(item)
             wakes: tuple[WakeRequest, ...] = ()
             if not _divergence_exists(state.divergences, "new-card", task, item, fingerprint):
diff --git a/python/helm_sync/tests/jq_parity.py b/python/helm_sync/tests/jq_parity.py
index 1348015dbdf..a6274f1d7b3 100644
--- a/python/helm_sync/tests/jq_parity.py
+++ b/python/helm_sync/tests/jq_parity.py
@@ -422,6 +422,7 @@ def _jq_plan_output(
     state: str = "queued",
     deleted_tsv: str = "",
     hold_kind: str = "",
+    tsv_existed: str = "true",
 ) -> bytes:
     """Run the production planner with one synthetic board item."""
     with tempfile.TemporaryDirectory(prefix="helm-jq-plan-") as temp:
@@ -504,7 +505,7 @@ def _jq_plan_output(
                 "1700000001",
                 "--arg",
                 "tsv_existed",
-                "true",
+                tsv_existed,
                 "--arg",
                 "retain_source",
                 "0",
@@ -807,6 +808,21 @@ def test_missing_card_never_before_seen_wakes_intake_without_closing(self) -> No
         jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, divergence_text)
         self.assertEqual(jq_output, python_output)
 
+    def test_missing_card_on_first_ever_sync_closes_instead_of_waking(self) -> None:
+        """Close an orphan card on a board's first-ever sync, when no identity cache existed yet."""
+        snapshot, wanted, state, raw_board, cards_tsv, divergence_text = _planner_fixture()
+        missing_task, missing_item, missing_node = "fixture-bootstrap-orphan-task", "PVTI_bootstrap_orphan", "DRAFT_bootstrap_orphan"
+        missing_card = _draft_card_snapshot(missing_item, missing_task, "Queued", missing_node)
+        extended_snapshot = BoardSnapshot(snapshot.board, snapshot.cards + (missing_card,), snapshot.fields)
+        raw_board["data"]["user"]["projectV2"]["items"]["nodes"].append(
+            _raw_draft_card(missing_item, missing_task, "Queued", missing_node)
+        )
+        state = SyncState(cards=state.cards, cache_existed=False)
+        plan = plan_board(extended_snapshot, (wanted,), state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, divergence_text, tsv_existed="false")
+        self.assertEqual(jq_output, python_output)
+
     def test_missing_card_already_done_is_left_alone(self) -> None:
         """Leave an already-Done orphaned card untouched, matching missing_entries' skip."""
         snapshot, wanted, state, raw_board, cards_tsv, divergence_text = _planner_fixture()

From b061c71efdba6428fc712303f51b77cf777bd5b5 Mon Sep 17 00:00:00 2001
From: firstmate-crewmate <brokentao@gmail.com>
Date: Fri, 18 Sep 2026 00:45:42 +0000
Subject: [PATCH 4/5] no-mistakes(review): docs(helm): fix stale note on
 implemented intake wake

---
 docs/1.architecture/3.helm-sync-python.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/docs/1.architecture/3.helm-sync-python.md b/docs/1.architecture/3.helm-sync-python.md
index accaf8f9c77..628f3c634c0 100644
--- a/docs/1.architecture/3.helm-sync-python.md
+++ b/docs/1.architecture/3.helm-sync-python.md
@@ -37,5 +37,6 @@ This slice does not read or write Helm state, call GitHub Projects, execute plan
 Those capabilities belong to later migration steps and must keep the current shell implementation active until their parity is proven.
 
 The planner now also closes a board card whose task has left every backlog (unless already Done) and holds rather than silently recreates a card that a captain deleted while its task is still live, matching `bin/fm-helm-lib.sh`'s `missing_entries` and `deleted_entries` phases at fixture scale.
-It does not yet raise jq's separate "captain added a card with no backlog task, run intake" wake, clear a stale dispatch marker on a closed card, or distinguish in-flight/blocked hold wording from queued.
+It also raises jq's separate "captain added a card with no backlog task, run intake" wake for a never-before-seen orphan task, and distinguishes a board's first-ever sync (no identity cache yet) from an ongoing one so it closes rather than wakes in that case, matching jq's `tsv_existed` check.
+It does not yet clear a stale dispatch marker on a closed card, or distinguish in-flight/blocked hold wording from queued.
 Project retention and the moved-card ledger (`bin/fm-helm-project-map.sh`) remain deferred to the follow-up state/executor migration slice, same as before.

From f8ff842f540592a0df74b96fc00257aad8e01215 Mon Sep 17 00:00:00 2001
From: firstmate-crewmate <brokentao@gmail.com>
Date: Fri, 18 Sep 2026 00:53:11 +0000
Subject: [PATCH 5/5] no-mistakes(review): fix(helm): guard empty-backlog
 missing phase, sync API docs

---
 docs/2.api/3.helm-sync-python.md    |  9 ++-
 python/helm_sync/planner.py         |  6 +-
 python/helm_sync/tests/jq_parity.py | 97 ++++++++++++++++++-----------
 3 files changed, 74 insertions(+), 38 deletions(-)

diff --git a/docs/2.api/3.helm-sync-python.md b/docs/2.api/3.helm-sync-python.md
index 50da518406d..7426967e0d2 100644
--- a/docs/2.api/3.helm-sync-python.md
+++ b/docs/2.api/3.helm-sync-python.md
@@ -181,6 +181,10 @@ Describes guarded field updates that preserve an Issue's title and body.
 
 Describes closing a previously synchronized card that is no longer in the desired set.
 
+### `WakeMissingCard(task, item, wakes=(), divergence_changes=())`
+
+Describes leaving a never-before-seen orphan card untouched while waking intake to review it.
+
 ### `SkipRecreatingDeletedCard(task, tombstone="", note="")`
 
 Describes declining to recreate a card whose cached item has left the board, leaving any captain hold or silent tombstone to the deleted-card phase instead.
@@ -340,7 +344,8 @@ The three-way comparison preserves captain edits, applies backlog edits that do
 
 ## Notes
 
-The planner returns typed actions, including `CreateDraft`, `UpdateDraft`, `UpdateIssueFields`, `NoChange`, `CloseMissingCard`, `SkipRecreatingDeletedCard`, `HoldDeletedTask`, and `KeepDeletedTombstone`.
+The planner returns typed actions, including `CreateDraft`, `UpdateDraft`, `UpdateIssueFields`, `NoChange`, `CloseMissingCard`, `WakeMissingCard`, `SkipRecreatingDeletedCard`, `HoldDeletedTask`, and `KeepDeletedTombstone`.
 It closes a board card whose task has no backlog record anywhere (unless already Done), and it holds rather than recreates a card whose task is still live but whose card a captain deleted.
-It does not yet raise the separate jq intake wake for a captain-created card with no matching task, nor clear a stale dispatch marker when it closes a missing card.
+It raises the separate jq intake wake for a captain-created card with no matching task, and it distinguishes a board's first-ever sync from an ongoing one so it closes rather than wakes when no identity cache exists yet.
+It does not yet clear a stale dispatch marker when it closes a missing card.
 The package has no state adapter, GraphQL client, executor, or command-line entry point in this migration slice.
diff --git a/python/helm_sync/planner.py b/python/helm_sync/planner.py
index da6270079f6..47a455ade0b 100644
--- a/python/helm_sync/planner.py
+++ b/python/helm_sync/planner.py
@@ -622,8 +622,12 @@ def _missing_phase_actions(
     port does not yet raise a note for), a valid but never-before-seen task id
     (a captain-created card with no backlog task) wakes intake instead of
     being closed when the identity cache already existed before this run,
-    and every other orphaned task is closed.
+    and every other orphaned task is closed. When this board's desired set is
+    empty, no card on the board is touched, matching jq's own guard against a
+    failed or empty backlog read mass-closing or mass-waking every card.
     """
+    if not desired_by_task:
+        return []
     done_option = _option(snapshot.fields, "Status", "Done")
     status_field = _field_id(snapshot.fields, "Status")
     actions: list[PlanAction] = []
diff --git a/python/helm_sync/tests/jq_parity.py b/python/helm_sync/tests/jq_parity.py
index a6274f1d7b3..7f86c10a151 100644
--- a/python/helm_sync/tests/jq_parity.py
+++ b/python/helm_sync/tests/jq_parity.py
@@ -415,7 +415,7 @@ def _planner_fixture(
 
 def _jq_plan_output(
     board: dict[str, object],
-    desired: DesiredCard,
+    desired: DesiredCard | None,
     cards_tsv: str,
     divergences_text: str,
     *,
@@ -424,7 +424,7 @@ def _jq_plan_output(
     hold_kind: str = "",
     tsv_existed: str = "true",
 ) -> bytes:
-    """Run the production planner with one synthetic board item."""
+    """Run the production planner with one synthetic board item, or none when `desired` is `None`."""
     with tempfile.TemporaryDirectory(prefix="helm-jq-plan-") as temp:
         root = Path(temp)
         board_path = root / "board.json"
@@ -435,43 +435,47 @@ def _jq_plan_output(
         divergences_path = root / "divergences.tsv"
         fps_path = root / "fps.tsv"
         board_path.write_bytes(_compact_json(board))
-        desired_record = {
-            "id": str(desired.task),
-            "state": state,
-            "home_path": "/fixture/main",
-            "repo": desired.repository or "",
-            "hold_kind": hold_kind,
-            "desired": {
-                "title": desired.title,
-                "body": desired.body,
-                "status": str(desired.status),
-                "kind": str(desired.kind),
-                "project": str(desired.project),
-                "priority_n": desired.priority_n,
-                "priority": str(desired.priority),
-                "board": {"owner": "fixture-owner", "number": 999},
-                "note": desired.note,
-            },
-        }
-        desired_path.write_bytes(_compact_json([desired_record]))
+        if desired is None:
+            desired_path.write_bytes(_compact_json([]))
+            fps_path.write_text("", encoding="utf-8")
+        else:
+            desired_record = {
+                "id": str(desired.task),
+                "state": state,
+                "home_path": "/fixture/main",
+                "repo": desired.repository or "",
+                "hold_kind": hold_kind,
+                "desired": {
+                    "title": desired.title,
+                    "body": desired.body,
+                    "status": str(desired.status),
+                    "kind": str(desired.kind),
+                    "project": str(desired.project),
+                    "priority_n": desired.priority_n,
+                    "priority": str(desired.priority),
+                    "board": {"owner": "fixture-owner", "number": 999},
+                    "note": desired.note,
+                },
+            }
+            desired_path.write_bytes(_compact_json([desired_record]))
+            body_hash = hashlib.sha256(desired.body.encode("utf-8")).hexdigest()
+            fingerprint = hashlib.sha256(
+                "\0".join(
+                    (
+                        str(desired.status),
+                        str(desired.priority),
+                        str(desired.project),
+                        str(desired.kind),
+                        desired.title,
+                        body_hash,
+                    )
+                ).encode("utf-8")
+            ).hexdigest()
+            fps_path.write_text(f"{desired.task}\t{fingerprint}\n", encoding="utf-8")
         cards_path.write_text(cards_tsv + "\n", encoding="utf-8")
         deleted_path.write_text(deleted_tsv, encoding="utf-8")
         markers_path.write_text("", encoding="utf-8")
         divergences_path.write_text(divergences_text, encoding="utf-8")
-        body_hash = hashlib.sha256(desired.body.encode("utf-8")).hexdigest()
-        fingerprint = hashlib.sha256(
-            "\0".join(
-                (
-                    str(desired.status),
-                    str(desired.priority),
-                    str(desired.project),
-                    str(desired.kind),
-                    desired.title,
-                    body_hash,
-                )
-            ).encode("utf-8")
-        ).hexdigest()
-        fps_path.write_text(f"{desired.task}\t{fingerprint}\n", encoding="utf-8")
         result = subprocess.run(
             [
                 "jq",
@@ -837,6 +841,29 @@ def test_missing_card_already_done_is_left_alone(self) -> None:
         jq_output = _jq_plan_output(raw_board, wanted, cards_tsv, divergence_text)
         self.assertEqual(jq_output, python_output)
 
+    def test_missing_phase_with_no_desired_records_touches_no_orphan_card(self) -> None:
+        """Guard missing_entries: leave every board card alone when this board's desired set is empty."""
+        board_ref = BoardRef(Owner("fixture-owner"), ProjectNumber(999))
+        schema = _all_fields()
+        orphan_task, orphan_item, orphan_node = "fixture-empty-backlog-orphan", "PVTI_empty_orphan", "DRAFT_empty_orphan"
+        orphan_card = _draft_card_snapshot(orphan_item, orphan_task, "Queued", orphan_node)
+        snapshot = BoardSnapshot(board_ref, (orphan_card,), schema)
+        state = SyncState()
+        raw_board = {
+            "data": {
+                "user": {
+                    "projectV2": {
+                        "fields": {"nodes": _raw_field_nodes()},
+                        "items": {"nodes": [_raw_draft_card(orphan_item, orphan_task, "Queued", orphan_node)]},
+                    }
+                }
+            }
+        }
+        plan = plan_board(snapshot, {}, state, force=True, epoch="1700000001")
+        python_output = _serialize_plan_for_jq_parity(plan)
+        jq_output = _jq_plan_output(raw_board, None, "", "")
+        self.assertEqual(jq_output, python_output)
+
     def test_deleted_card_for_a_live_task_raises_a_captain_hold(self) -> None:
         """Preserve a captain's card deletion and raise one hold instead of recreating it."""
         wanted, baseline, snapshot, cards_tsv = _deleted_task_fixture(