feat(cancel): STOP-status task cancellation + relaunch-hole close (ORCH-090)

Introduce the dedicated Plane STOP status as a single declarative task-cancel
mechanism: stop the active agent (graceful SIGTERM cascade), cancel all jobs
(terminal `cancelled`, never requeued), remove the worktree + delete the remote
feature branch (never main, never force-push), drive the task to the new
system-terminal state `cancelled` and tombstone the natural keys so a later
"To Analyse" re-creates it from scratch (docs artefacts preserved). STOP during a
critical merge/deploy window is deferred until the irreversible step finishes
honestly. Also closes the relaunch hole: handle_status_start relaunch is gated to
the `analysis` stage; the only pipeline-start entry point remains "To Analyse".

Cross-cutting (adr-0026): the "task terminal" predicate is widened {done} ->
{done, cancelled} in serial_gate / task_deps / stages sink + reaper/worker
requeue guards. STAGE_TRANSITIONS exit-gates / QG_CHECKS / check_* are unchanged
(`cancelled` is a sink, not a new edge). Additive, never-raise, restart-safe,
under kill-switch ORCH_STOP_STATUS_ENABLED (off -> zero regression).

New: src/cancel.py (leaf), src/gitea.py (delete_remote_branch), tasks columns
cancelled_at/cancel_requested_at, jobs status `cancelled`, GET /queue `stop` block.
Tests: tests/test_stop_status.py (TC-01..TC-14 + D7); full suite green (1345).
Docs updated in-PR (architecture README, CLAUDE.md, README.md, .env.example,
CHANGELOG). ADR-001 D4 refinement: plane_issue_id is tombstoned too (the lookup
ORs on it) — original UUID recoverable from the parseable suffix.

Refs: ORCH-090

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/test_auto_labels_invariants.py

@@ -13,9 +13,10 @@ os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
 
 def test_tc26_stage_transitions_unchanged():
     from src.stages import STAGE_TRANSITIONS
+    # ORCH-090 (adr-0026): `cancelled` terminal sink added (parallel to `done`).
     assert set(STAGE_TRANSITIONS) == {
         "created", "analysis", "architecture", "development", "review",
-        "testing", "deploy-staging", "deploy", "done",
+        "testing", "deploy-staging", "deploy", "done", "cancelled",
     }
     # The two human gates still use their existing QG names (unchanged).
     assert STAGE_TRANSITIONS["analysis"]["qg"] == "check_analysis_approved"

tests/test_config.py

@@ -219,11 +219,15 @@ def test_reaper_settings_env_override(monkeypatch):
 # check_branch_mergeable signature is intact (AC-13).
 # ---------------------------------------------------------------------------
 def test_tc19_stage_transitions_unchanged():
-    """No new pipeline stage was introduced by ORCH-065."""
+    """No new pipeline EDGE was introduced by ORCH-065.
+
+    ORCH-090 (adr-0026) adds `cancelled` as a terminal SINK (parallel to `done`),
+    which is not a new edge — no exit-gate of any edge changed.
+    """
     from src.stages import STAGE_TRANSITIONS
     assert set(STAGE_TRANSITIONS) == {
         "created", "analysis", "architecture", "development", "review",
-        "testing", "deploy-staging", "deploy", "done",
+        "testing", "deploy-staging", "deploy", "done", "cancelled",
     }
 
 

tests/test_plane_status_model.py

@@ -125,6 +125,8 @@ def test_tc22_stage_transitions_unchanged():
         "deploy-staging": {"next": "deploy", "agent": "deployer", "qg": "check_staging_status"},
         "deploy": {"next": "done", "agent": None, "qg": "check_deploy_status"},
         "done": {"next": None, "agent": None, "qg": None},
+        # ORCH-090 (adr-0026): terminal SINK for a STOP-cancelled task.
+        "cancelled": {"next": None, "agent": None, "qg": None},
     }
 
 

tests/test_qg_registry_snapshot.py

@@ -56,6 +56,9 @@ _EXPECTED_TRANSITIONS = {
     "deploy-staging": {"next": "deploy",          "agent": "deployer",  "qg": "check_staging_status"},
     "deploy":         {"next": "done",            "agent": None,        "qg": "check_deploy_status"},
     "done":           {"next": None,              "agent": None,        "qg": None},
+    # ORCH-090 (adr-0026): terminal SINK for a STOP-cancelled task (parallel to
+    # `done`; not a new edge — no exit-gate changed).
+    "cancelled":      {"next": None,              "agent": None,        "qg": None},
 }
 
 

tests/test_serial_gate.py

@@ -180,9 +180,11 @@ def test_snapshot_shape_and_never_raises(monkeypatch):
 def test_registries_unchanged():
     from src.stages import STAGE_TRANSITIONS
     from src.qg.checks import QG_CHECKS
+    # ORCH-090 (adr-0026): `cancelled` is added as a terminal SINK (parallel to
+    # `done`), NOT a new pipeline edge — serial-gate FIFO semantics are unchanged.
     assert set(STAGE_TRANSITIONS) == {
         "created", "analysis", "architecture", "development", "review",
-        "testing", "deploy-staging", "deploy", "done",
+        "testing", "deploy-staging", "deploy", "done", "cancelled",
     }
     # No serial-gate QG check was introduced (the gate is a scheduler condition).
     assert not any("serial" in k for k in QG_CHECKS), "no new QG check expected"

tests/test_stages_invariants.py

@@ -39,6 +39,9 @@ _EXPECTED_TRANSITIONS = {
     "deploy-staging": {"next": "deploy",          "agent": "deployer",  "qg": "check_staging_status"},
     "deploy":         {"next": "done",            "agent": None,        "qg": "check_deploy_status"},
     "done":           {"next": None,              "agent": None,        "qg": None},
+    # ORCH-090 (adr-0026): terminal SINK for a STOP-cancelled task (parallel to
+    # `done`; not a new edge — no exit-gate changed).
+    "cancelled":      {"next": None,              "agent": None,        "qg": None},
 }
 
 

tests/test_stop_status.py

@@ -0,0 +1,454 @@
+"""ORCH-090 — STOP-status task cancellation + relaunch-hole close (unit + integ).
+
+Covers 04-test-plan.yaml TC-01..TC-14 + the ADR-001 D7 deferred-cancel path:
+
+  TC-01  STOP recognised + routed to handle_stop; unknown task -> no-op, never-raise.
+  TC-02  active agent stopped via launcher.stop_process by jobs.pid; idle -> no-op.
+  TC-03  queued+running jobs of the task -> terminal 'cancelled'; claim skips them.
+  TC-04  reaper does NOT requeue a job of a terminal (cancelled) task.
+  TC-05  full reset: remove_worktree + delete_remote_branch called; main untouched.
+  TC-06  docs artefacts (and the task row) survive the reset.
+  TC-07  idempotency: STOP on cancelled / done / missing -> no-op, no exception.
+  TC-08  kill-switch off -> STOP inert; relaunch-hole gate inert.
+  TC-09  GET /queue carries a read-only `stop` block; never-raise.
+  TC-10  relaunch-hole closed: manual To Analyse on a mid-pipeline task -> no job.
+  TC-11  To Analyse on analysis (idle) relaunches analyst; new task -> start_pipeline.
+  TC-12  terminal-skip / restart-safe: reconciler skips a cancelled task; cancelled
+         jobs are not revived by requeue_running_jobs.
+  TC-13  e2e STOP: agent stopped, jobs cancelled, branch/worktree removed, durable
+         'cancelled', keys tombstoned, notifications fired.
+  TC-14  additive DB migration is idempotent (re-init_db) + columns present.
+  D7     STOP in a critical merge/deploy window is DEFERRED, then applied by the
+         deploy finalizer.
+"""
+import os
+import tempfile
+
+import pytest
+
+os.environ["ORCH_DB_PATH"] = os.path.join(tempfile.gettempdir(), "test_stop_status.db")
+os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
+os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
+
+import src.db as db  # noqa: E402
+from src.db import (  # noqa: E402
+    init_db, get_db, claim_next_job, get_task,
+    cancel_jobs_for_task, mark_task_cancelled, get_task_by_plane_id,
+    requeue_running_jobs, get_job,
+)
+from src import config as cfg  # noqa: E402
+from src import cancel as cancel_mod  # noqa: E402
+from src import stage_engine  # noqa: E402
+
+
+@pytest.fixture(autouse=True)
+def fresh_db(tmp_path, monkeypatch):
+    dbfile = tmp_path / "stop.db"
+    monkeypatch.setattr(db.settings, "db_path", str(dbfile))
+    # STOP feature ON, all repos. Isolate repos_dir so the critical-window probe
+    # (deploy markers / merge-lease) sees a clean tree by default.
+    monkeypatch.setattr(cfg.settings, "stop_status_enabled", True, raising=False)
+    monkeypatch.setattr(cfg.settings, "stop_status_repos", "", raising=False)
+    monkeypatch.setattr(cfg.settings, "repos_dir", str(tmp_path / "repos"), raising=False)
+    monkeypatch.setattr(cfg.settings, "host_repos_dir", str(tmp_path / "repos"), raising=False)
+    monkeypatch.setattr(cfg.settings, "serial_gate_enabled", False, raising=False)
+    monkeypatch.setattr(cfg.settings, "task_deps_enabled", False, raising=False)
+    # Silence network side effects of cancel notifications.
+    monkeypatch.setattr("src.stage_engine.plane_add_comment", lambda *a, **k: None, raising=False)
+    monkeypatch.setattr("src.notifications.update_task_tracker", lambda *a, **k: None, raising=False)
+    init_db()
+    yield
+
+
+# --------------------------------------------------------------------------- helpers
+def _make_task(plane_id, work_item_id, stage="development", repo="orchestrator",
+               branch=None):
+    branch = branch or f"feature/{work_item_id}-slug"
+    conn = get_db()
+    cur = conn.execute(
+        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, plane_issue_id, title) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?)",
+        (plane_id, work_item_id, repo, branch, stage, plane_id, work_item_id),
+    )
+    tid = cur.lastrowid
+    conn.commit()
+    conn.close()
+    return tid
+
+
+def _make_job(task_id, repo="orchestrator", agent="developer", status="running",
+              pid=None, run_id=None, attempts=1, max_attempts=2):
+    conn = get_db()
+    cur = conn.execute(
+        "INSERT INTO jobs (agent, repo, task_id, status, pid, run_id, attempts, max_attempts) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        (agent, repo, task_id, status, pid, run_id, attempts, max_attempts),
+    )
+    jid = cur.lastrowid
+    conn.commit()
+    conn.close()
+    return jid
+
+
+def _job_status(job_id):
+    j = get_job(job_id)
+    return j["status"] if j else None
+
+
+def _stub_full_reset(monkeypatch):
+    """Stub the side-effecting cleanup steps (signals / git / gitea) of a full reset."""
+    calls = {"stop": [], "worktree": [], "branch": []}
+    from src.agents.launcher import launcher
+
+    def _stop(pid, run_id, *, reason="stop"):
+        calls["stop"].append((pid, run_id, reason))
+        return True
+    monkeypatch.setattr(launcher, "stop_process", _stop, raising=True)
+    monkeypatch.setattr("src.git_worktree.remove_worktree",
+                        lambda repo, branch: calls["worktree"].append((repo, branch)),
+                        raising=True)
+    monkeypatch.setattr("src.gitea.delete_remote_branch",
+                        lambda repo, branch: calls["branch"].append((repo, branch)) or True,
+                        raising=True)
+    return calls
+
+
+# =========================================================================== TC-01
+@pytest.mark.asyncio
+async def test_tc01_stop_routed_and_unknown_is_noop(monkeypatch):
+    from src.webhooks import plane as plane_wh
+
+    proj_states = {
+        "stop": "STOP-UUID", "to_analyse": "TA-UUID", "approved": "AP-UUID",
+        "rejected": "RJ-UUID", "confirm_deploy": None,
+    }
+    monkeypatch.setattr("src.plane_sync.get_project_states", lambda pid: proj_states)
+    seen = []
+
+    async def _stub_stop(data, project_id=""):
+        seen.append(data.get("id"))
+    monkeypatch.setattr(plane_wh, "handle_stop", _stub_stop)
+
+    # STOP state -> routed to handle_stop.
+    await plane_wh.handle_issue_updated({"id": "PL-1", "state": {"id": "STOP-UUID"}}, "proj")
+    assert seen == ["PL-1"]
+
+    # A non-STOP state does not route to handle_stop.
+    await plane_wh.handle_issue_updated({"id": "PL-2", "state": {"id": "AP-UUID"}}, "proj")
+    assert seen == ["PL-1"]
+
+    # Unknown task on the real handler -> no-op, never raises.
+    await plane_wh.handle_stop({"id": "does-not-exist"}, "proj")
+
+
+# =========================================================================== TC-02
+def test_tc02_stop_active_agent_by_pid(monkeypatch):
+    calls = _stub_full_reset(monkeypatch)
+    tid = _make_task("PL-10", "ORCH-310", stage="development")
+    _make_job(tid, status="running", pid=4242, run_id=77)
+
+    res = stage_engine.cancel_task(tid)
+    assert res["ok"] and not res["deferred"]
+    assert calls["stop"] == [(4242, 77, f"STOP cancel task {tid}")]
+    assert res["stopped"] == 1
+
+
+def test_tc02_idle_agent_no_stop(monkeypatch):
+    calls = _stub_full_reset(monkeypatch)
+    tid = _make_task("PL-11", "ORCH-311", stage="development")
+    _make_job(tid, status="queued", pid=None)  # no running process
+
+    res = stage_engine.cancel_task(tid)
+    assert res["ok"] and res["stopped"] == 0
+    assert calls["stop"] == []
+
+
+# =========================================================================== TC-03
+def test_tc03_jobs_cancelled_and_claim_skips(monkeypatch):
+    _stub_full_reset(monkeypatch)
+    tid = _make_task("PL-20", "ORCH-320", stage="development")
+    jq = _make_job(tid, status="queued")
+    jr = _make_job(tid, status="running", pid=None)
+
+    stage_engine.cancel_task(tid)
+    assert _job_status(jq) == "cancelled"
+    assert _job_status(jr) == "cancelled"
+    # claim_next_job selects only status='queued' -> a cancelled job is never claimed.
+    assert claim_next_job() is None
+
+
+def test_tc03_cancel_jobs_helper_only_queued(monkeypatch):
+    tid = _make_task("PL-21", "ORCH-321")
+    jq = _make_job(tid, status="queued")
+    jr = _make_job(tid, status="running", pid=None)
+    n = cancel_jobs_for_task(tid, only_queued=True)
+    assert n == 1
+    assert _job_status(jq) == "cancelled"
+    assert _job_status(jr) == "running"  # the running deploy/merge actor is left alone
+
+
+# =========================================================================== TC-04
+def test_tc04_reaper_does_not_requeue_terminal_task(monkeypatch):
+    from src.job_reaper import JobReaper
+    tid = _make_task("PL-30", "ORCH-330", stage="development")
+    jid = _make_job(tid, status="running", pid=999999, attempts=1, max_attempts=2)
+    # Task is flipped to cancelled (as STOP would) while the job is still running.
+    mark_task_cancelled(tid)
+
+    reaper = JobReaper()
+    job = get_job(jid)
+    reaper._reap_unknown_outcome(job, reason="dead pid")
+    # NOT requeued (attempts<max would normally requeue) -> terminal 'cancelled'.
+    assert _job_status(jid) == "cancelled"
+
+
+# =========================================================================== TC-05
+def test_tc05_full_reset_removes_branch_and_worktree(monkeypatch):
+    calls = _stub_full_reset(monkeypatch)
+    tid = _make_task("PL-40", "ORCH-340", stage="review", branch="feature/ORCH-340-x")
+
+    stage_engine.cancel_task(tid)
+    assert calls["worktree"] == [("orchestrator", "feature/ORCH-340-x")]
+    assert calls["branch"] == [("orchestrator", "feature/ORCH-340-x")]
+
+
+def test_tc05_delete_remote_branch_refuses_main():
+    from src import gitea
+    # main is never deletable by the cancel path (self-hosting safety, NFR-3).
+    assert gitea.delete_remote_branch("orchestrator", "main") is False
+    assert gitea.delete_remote_branch("orchestrator", "master") is False
+
+
+# =========================================================================== TC-06
+def test_tc06_docs_and_task_row_survive(monkeypatch, tmp_path):
+    _stub_full_reset(monkeypatch)
+    tid = _make_task("PL-50", "ORCH-350", stage="development")
+    # A stand-in docs artefact: cancel must not delete it.
+    docs = tmp_path / "docs" / "work-items" / "ORCH-350"
+    docs.mkdir(parents=True)
+    (docs / "02-trz.md").write_text("trz")
+
+    stage_engine.cancel_task(tid)
+    assert (docs / "02-trz.md").exists(), "docs artefacts must be preserved"
+    # The task ROW is kept (durable audit), flipped to cancelled.
+    assert get_task(tid)["stage"] == "cancelled"
+
+
+# =========================================================================== TC-07
+def test_tc07_idempotent_on_cancelled_done_missing(monkeypatch):
+    calls = _stub_full_reset(monkeypatch)
+    # already cancelled
+    tid = _make_task("PL-60", "ORCH-360", stage="cancelled")
+    res = stage_engine.cancel_task(tid)
+    assert res["ok"] and res["note"].startswith("already-terminal")
+    assert calls["stop"] == [] and calls["branch"] == []
+    # done
+    tid2 = _make_task("PL-61", "ORCH-361", stage="done")
+    res2 = stage_engine.cancel_task(tid2)
+    assert res2["note"].startswith("already-terminal")
+    # missing
+    res3 = stage_engine.cancel_task(999999)
+    assert res3["note"] == "no-task"
+
+
+# =========================================================================== TC-08
+def test_tc08_kill_switch_off_inert(monkeypatch):
+    monkeypatch.setattr(cfg.settings, "stop_status_enabled", False, raising=False)
+    assert cancel_mod.applies("orchestrator") is False
+
+
+@pytest.mark.asyncio
+async def test_tc08_kill_switch_off_handle_stop_noop(monkeypatch):
+    monkeypatch.setattr(cfg.settings, "stop_status_enabled", False, raising=False)
+    calls = _stub_full_reset(monkeypatch)
+    from src.webhooks import plane as plane_wh
+    tid = _make_task("PL-70", "ORCH-370", stage="development")
+    _make_job(tid, status="running", pid=4242)
+    await plane_wh.handle_stop({"id": "PL-70"}, "proj")
+    # Nothing was cancelled (kill-switch off -> applies() False -> no-op).
+    assert calls["stop"] == []
+    assert get_task(tid)["stage"] == "development"
+
+
+def test_tc08_scope_csv(monkeypatch):
+    monkeypatch.setattr(cfg.settings, "stop_status_repos", "enduro-trails", raising=False)
+    assert cancel_mod.applies("enduro-trails") is True
+    assert cancel_mod.applies("orchestrator") is False
+
+
+# =========================================================================== TC-09
+def test_tc09_queue_has_stop_block_and_keeps_keys(monkeypatch):
+    import asyncio
+    from src import main
+    payload = asyncio.run(main.queue())
+    for key in ("counts", "serial_gate", "task_deps", "auto_labels", "recent"):
+        assert key in payload, f"existing /queue key '{key}' preserved"
+    assert "stop" in payload
+    blk = payload["stop"]
+    assert blk["enabled"] is True
+    assert "repos" in blk and "cancelled_count" in blk and "recent" in blk
+
+
+def test_tc09_snapshot_never_raises(monkeypatch):
+    # Force a DB error inside the snapshot -> minimal dict, no raise.
+    monkeypatch.setattr("src.db.cancelled_tasks_snapshot",
+                        lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")))
+    snap = cancel_mod.snapshot()
+    assert snap["enabled"] is True and snap["cancelled_count"] == 0
+
+
+# =========================================================================== TC-10
+@pytest.mark.asyncio
+async def test_tc10_relaunch_hole_closed_midpipeline(monkeypatch):
+    from src.webhooks import plane as plane_wh
+    monkeypatch.setattr("src.plane_sync.add_comment", lambda *a, **k: None, raising=False)
+    monkeypatch.setattr("src.plane_sync.set_issue_analysis", lambda *a, **k: None, raising=False)
+    tid = _make_task("PL-80", "ORCH-380", stage="development")
+
+    await plane_wh.handle_status_start({"id": "PL-80"}, "proj")
+    # No stage agent was relaunched (no job created) for a mid-pipeline task.
+    conn = get_db()
+    n = conn.execute("SELECT COUNT(*) FROM jobs WHERE task_id=?", (tid,)).fetchone()[0]
+    conn.close()
+    assert n == 0
+
+
+# =========================================================================== TC-11
+@pytest.mark.asyncio
+async def test_tc11_analysis_idle_relaunches_analyst(monkeypatch):
+    from src.webhooks import plane as plane_wh
+    monkeypatch.setattr("src.plane_sync.add_comment", lambda *a, **k: None, raising=False)
+    monkeypatch.setattr("src.plane_sync.set_issue_analysis", lambda *a, **k: None, raising=False)
+    tid = _make_task("PL-90", "ORCH-390", stage="analysis")
+
+    await plane_wh.handle_status_start({"id": "PL-90"}, "proj")
+    conn = get_db()
+    rows = conn.execute("SELECT agent FROM jobs WHERE task_id=?", (tid,)).fetchall()
+    conn.close()
+    assert [r[0] for r in rows] == ["analyst"], "analyst resume is still legitimate"
+
+
+@pytest.mark.asyncio
+async def test_tc11_new_task_starts_pipeline(monkeypatch):
+    from src.webhooks import plane as plane_wh
+    started = []
+
+    async def _stub_start(data, project_id=""):
+        started.append(data.get("id"))
+    monkeypatch.setattr(plane_wh, "start_pipeline", _stub_start)
+    await plane_wh.handle_status_start({"id": "PL-NEW"}, "proj")
+    assert started == ["PL-NEW"]  # the ONLY pipeline-start entry point
+
+
+# =========================================================================== TC-12
+def test_tc12_reconciler_skips_cancelled(monkeypatch):
+    from src.reconciler import Reconciler
+    # Avoid any Plane network in the gate pass.
+    monkeypatch.setattr("src.reconciler.fetch_issue_state",
+                        lambda *a, **k: (_ for _ in ()).throw(AssertionError("no net")),
+                        raising=False)
+    tid = _make_task("PL-100", "ORCH-400", stage="development")
+    mark_task_cancelled(tid)
+    rec = Reconciler()
+    rec.reconcile_gate_once()
+    assert rec.skipped_terminal_total == 1
+
+
+def test_tc12_requeue_running_does_not_revive_cancelled(monkeypatch):
+    tid = _make_task("PL-101", "ORCH-401", stage="development")
+    jc = _make_job(tid, status="running", pid=None)
+    cancel_jobs_for_task(tid)  # -> cancelled
+    assert _job_status(jc) == "cancelled"
+    # Startup recovery flips only 'running' jobs; a cancelled job is untouched.
+    requeue_running_jobs()
+    assert _job_status(jc) == "cancelled"
+
+
+# =========================================================================== TC-13
+def test_tc13_end_to_end_stop(monkeypatch):
+    calls = _stub_full_reset(monkeypatch)
+    tid = _make_task("PL-110", "ORCH-410", stage="review", branch="feature/ORCH-410-e2e")
+    jr = _make_job(tid, status="running", pid=5555, run_id=11)
+    jq = _make_job(tid, status="queued")
+
+    res = stage_engine.cancel_task(tid, reason="Plane STOP status")
+    assert res["ok"] and not res["deferred"]
+    # agent stopped
+    assert calls["stop"] and calls["stop"][0][0] == 5555
+    # jobs cancelled
+    assert _job_status(jr) == "cancelled" and _job_status(jq) == "cancelled"
+    # worktree + branch removed
+    assert calls["worktree"] and calls["branch"]
+    # durable terminal + key tombstone (re-create via To Analyse no longer collides)
+    t = get_task(tid)
+    assert t["stage"] == "cancelled" and t["cancelled_at"]
+    assert t["plane_id"].endswith(f"#cancelled-{tid}")
+    assert t["work_item_id"].endswith(f"#cancelled-{tid}")
+    # plane_issue_id is tombstoned too (the lookup ORs on it) but the original UUID
+    # remains recoverable from the parseable suffix (audit link preserved).
+    assert t["plane_issue_id"] == f"PL-110#cancelled-{tid}"
+    assert t["plane_issue_id"].split("#cancelled-")[0] == "PL-110"
+    assert get_task_by_plane_id("PL-110") is None  # freed for a fresh start
+
+
+# =========================================================================== TC-14
+def test_tc14_migration_idempotent_and_columns_present():
+    # Re-running init_db must not fail (idempotent _ensure_column).
+    init_db()
+    init_db()
+    conn = get_db()
+    cols = {r[1] for r in conn.execute("PRAGMA table_info(tasks)").fetchall()}
+    conn.close()
+    assert "cancelled_at" in cols and "cancel_requested_at" in cols
+
+
+def test_tc14_existing_contracts_intact():
+    # The additive job status set still has the original statuses working.
+    tid = _make_task("PL-120", "ORCH-420")
+    jid = _make_job(tid, status="queued")
+    # A queued job is still claimable when no gate blocks it.
+    claimed = claim_next_job()
+    assert claimed is not None and claimed["id"] == jid
+
+
+# =========================================================================== D7
+def test_d7_stop_in_critical_window_defers(monkeypatch):
+    calls = _stub_full_reset(monkeypatch)
+    from src import self_deploy
+    tid = _make_task("PL-130", "ORCH-430", stage="deploy", branch="feature/ORCH-430-d")
+    # self-deploy Phase B initiated -> critical window.
+    self_deploy.write_marker("orchestrator", "ORCH-430", self_deploy.INITIATED, content="1")
+    jq = _make_job(tid, status="queued")
+    jr = _make_job(tid, status="running", pid=7777)  # the deploy actor
+
+    res = stage_engine.cancel_task(tid)
+    assert res["deferred"] is True and res["ok"]
+    # Only queued jobs cancelled; the running deploy actor is NOT killed.
+    assert _job_status(jq) == "cancelled"
+    assert _job_status(jr) == "running"
+    assert calls["stop"] == [] and calls["branch"] == []
+    # The deferred flag is durable; the task is NOT yet terminal.
+    t = get_task(tid)
+    assert t["cancel_requested_at"] and t["stage"] == "deploy"
+
+
+def test_d7_in_critical_window_detection(monkeypatch):
+    from src import self_deploy
+    task = {"repo": "orchestrator", "work_item_id": "ORCH-431", "branch": "feature/x"}
+    assert cancel_mod.in_critical_window(task) is False
+    self_deploy.write_marker("orchestrator", "ORCH-431", self_deploy.INITIATED, content="1")
+    assert cancel_mod.in_critical_window(task) is True
+
+
+def test_d7_deferred_applied_by_finalizer(monkeypatch):
+    """After the irreversible step finishes, the finalizer applies the deferred cancel."""
+    calls = _stub_full_reset(monkeypatch)
+    tid = _make_task("PL-140", "ORCH-440", stage="development", branch="feature/ORCH-440-d")
+    # Mark a deferred cancellation pending (as the critical-window path would).
+    db.set_task_cancel_requested(tid)
+
+    # force=True is what run_deploy_finalizer uses once the step completed honestly.
+    res = stage_engine.cancel_task(tid, force=True, source="deferred")
+    assert res["ok"] and not res["deferred"]
+    assert get_task(tid)["stage"] == "cancelled"
+    assert calls["branch"], "deferred cancel applies the full reset"