fix(helm): add missing and deleted-card phases to Python sync planner

docs/1.architecture/3.helm-sync-python.md

@@ -36,4 +36,7 @@ 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 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.

docs/2.api/3.helm-sync-python.md

@@ -181,6 +181,14 @@ 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.
+
 ### `RecordDispatchRequest(task, item, option, fingerprint)`
 
 Describes recording a new captain dispatch request.
@@ -193,7 +201,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 +344,8 @@ 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`, `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 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.

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,
     )
 
 

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)
@@ -307,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)))
@@ -451,6 +453,25 @@ 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."""
+
+    task: TaskId
+    tombstone: str = ""
+    note: str = ""
+
+
 @dataclass(frozen=True)
 class RecordDispatchRequest:
     """Describe persisting a new captain dispatch request."""
@@ -482,8 +503,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 +547,8 @@ class ForgetDivergence:
     | UpdateDraft
     | UpdateIssueFields
     | CloseMissingCard
+    | WakeMissingCard
+    | SkipRecreatingDeletedCard
     | RecordDispatchRequest
     | ClearDispatchRequest
     | WritePriorityToBacklog

python/helm_sync/planner.py

@@ -14,6 +14,7 @@
     BoardSnapshot,
     CardBaseline,
     CardSnapshot,
+    CloseMissingCard,
     CreateDraft,
     DesiredCard,
     DivergenceChange,
@@ -22,15 +23,19 @@
     FieldWrite,
     FieldName,
     Fingerprint,
+    HoldDeletedTask,
     ItemId,
+    KeepDeletedTombstone,
     NoChange,
     OptionId,
     PlanAction,
     Signature,
+    SkipRecreatingDeletedCard,
     SyncState,
     TaskId,
     UpdateDraft,
     UpdateIssueFields,
+    WakeMissingCard,
     WakeRequest,
     parse_item_id,
     parse_task_id,
@@ -607,6 +612,121 @@ def _unsupported_repo_note(
     return "", ()
 
 
+def _missing_phase_actions(
+    snapshot: BoardSnapshot, desired_by_task: Mapping[TaskId, DesiredCard], state: SyncState
+) -> list[PlanAction]:
+    """Close or wake on previously seen and never-before-seen orphan cards.
+
+    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 when the identity cache already existed before this run,
+    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] = []
+    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
+        item = parse_item_id(card.item)
+        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):
+                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=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 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}
+    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" or wanted.hold_kind == "captain":
+            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 +764,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 +828,9 @@ def plan_board(
             )
         )
 
+    actions.extend(_missing_phase_actions(snapshot, desired_by_task, state))
+    actions.extend(_deleted_phase_actions(snapshot, desired_by_task, state))
+
     canonical = [
         {
             "item": str(card.item),
@@ -718,6 +851,16 @@ 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, WakeMissingCard):
+        return _legacy_missing_wake_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 +971,63 @@ 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_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 = (
+        "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)