fix(api): reject stale ticket updates and completion atomically

planfile/__init__.py

@@ -419,6 +419,7 @@ def complete_ticket(
         *,
         reason: str | None = None,
         actor: str | None = None,
+        expected_updated_at: str | None = None,
     ) -> Ticket | None:
         ticket = self.get_ticket(ticket_id)
         if not ticket:
@@ -446,7 +447,10 @@ def complete_ticket(
             last_error=None,
         )
         execution = TicketExecution(**execution_data)
-        return self.update_ticket(ticket_id, status="done", execution=execution, outputs=outputs, reason=reason, actor=actor)
+        return self.update_ticket(
+            ticket_id, status="done", execution=execution, outputs=outputs,
+            reason=reason, actor=actor, expected_updated_at=expected_updated_at,
+        )
 
     def fail_ticket(
         self,

planfile/api/server.py

@@ -80,7 +80,11 @@ async def lifespan(_: FastAPI):
     lifespan=lifespan,
 )
 
-API_CAPABILITIES = ["ticket.fail.expected_updated_at"]
+API_CAPABILITIES = [
+    "ticket.fail.expected_updated_at",
+    "ticket.update.expected_updated_at",
+    "ticket.complete.expected_updated_at",
+]
 
 
 @app.exception_handler(ImmutableTerminalReopenError)
@@ -153,6 +157,10 @@ class TicketUpdate(BaseModel):
     actor: str | None = None
 
 
+class TicketUpdateIfCurrentRequest(TicketUpdate):
+    expected_updated_at: str = Field(..., min_length=1, pattern=r"\S")
+
+
 class TicketEvidenceAppendRequest(BaseModel):
     """Atomic, retry-safe evidence append.
 
@@ -211,6 +219,10 @@ class TicketCompleteRequest(BaseModel):
     actor: str | None = None
 
 
+class TicketCompleteIfCurrentRequest(TicketCompleteRequest):
+    expected_updated_at: str = Field(..., min_length=1, pattern=r"\S")
+
+
 class TicketFailRequest(BaseModel):
     error: str
     reason: str | None = None
@@ -1000,6 +1012,13 @@ async def update_ticket(ticket_id: str, body: TicketUpdate):
     return ticket.model_dump(mode="json", exclude_none=True)
 
 
+@app.post("/tickets/{ticket_id}/update-if-current", tags=["tickets"])
+async def update_ticket_if_current(ticket_id: str, body: TicketUpdateIfCurrentRequest):
+    # A separate route prevents older servers silently ignoring the guard.
+    # Keep validation, the store mutation lock and event delivery shared.
+    return await update_ticket(ticket_id, body)
+
+
 @app.post("/tickets/{ticket_id}/evidence", tags=["tickets"])
 def append_ticket_evidence(
     ticket_id: str,
@@ -1135,13 +1154,19 @@ async def complete_ticket(ticket_id: str, body: TicketCompleteRequest):
         completion_receipt=body.completion_receipt,
         reason=body.reason or (body.completion_receipt or {}).get("reason") or body.note or "ticket_completed_via_api",
         actor=body.actor or (body.completion_receipt or {}).get("actor") or "unknown:api",
+        expected_updated_at=getattr(body, "expected_updated_at", None),
     )
     if not ticket:
         raise HTTPException(404, f"Ticket {ticket_id} not found")
     await _broadcast_ticket_event("ticket.execution.changed", "complete", ticket)
     return ticket.model_dump(mode="json", exclude_none=True)
 
 
+@app.post("/tickets/{ticket_id}/complete-if-current", tags=["tickets"])
+async def complete_ticket_if_current(ticket_id: str, body: TicketCompleteIfCurrentRequest):
+    return await complete_ticket(ticket_id, body)
+
+
 async def _fail_ticket(ticket_id: str, body: TicketFailRequest):
     pf = get_planfile()
     current = pf.get_ticket(ticket_id, repair_index=False)

tests/test_ticket_update_precondition.py

@@ -0,0 +1,104 @@
+"""Conditional observer writes cannot replace newer worker evidence."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from planfile import Planfile, TicketExecution, TicketOutputs
+from planfile.api import server
+
+
+@pytest.fixture(params=["yaml", "sharded-yaml"])
+def context(tmp_path, monkeypatch, request):
+    pf = Planfile(str(tmp_path))
+    if request.param == "sharded-yaml":
+        pf.store.migrate_to_sharded_yaml(shard_size=100)
+    ticket = pf.create_ticket(
+        name="Publication observation",
+        execution=TicketExecution(state="ready", max_attempts=2),
+    )
+    monkeypatch.setattr(server, "get_planfile", lambda: pf)
+    return pf, ticket, TestClient(server.app)
+
+
+@pytest.mark.parametrize("operation", ["update", "complete"])
+@pytest.mark.parametrize("revision", [None, "", "   "])
+def test_conditional_write_requires_revision(context, operation, revision):
+    pf, ticket, client = context
+    body = {} if revision is None else {"expected_updated_at": revision}
+    response = client.post(f"/tickets/{ticket.id}/{operation}-if-current", json=body)
+    assert response.status_code == 422
+    assert pf.get_ticket(ticket.id).updated_at == ticket.updated_at
+
+
+@pytest.mark.parametrize("operation", ["update", "complete"])
+def test_conditional_write_accepts_current_revision(context, operation):
+    pf, ticket, client = context
+    body = {"expected_updated_at": ticket.model_dump(mode="json")["updated_at"]}
+    body.update({"priority": "high"} if operation == "update" else {"result": {"observed": True}})
+    response = client.post(f"/tickets/{ticket.id}/{operation}-if-current", json=body)
+    assert response.status_code == 200
+    current = pf.get_ticket(ticket.id)
+    assert current.updated_at != ticket.updated_at
+    if operation == "update":
+        assert current.priority == "high"
+        assert current.execution.state == "ready"
+    else:
+        assert current.status == "done"
+        assert current.outputs.result == {"observed": True}
+    assert f"ticket.{operation}.expected_updated_at" in client.get("/health").json()["capabilities"]
+
+
+@pytest.mark.parametrize("operation", ["update", "complete"])
+@pytest.mark.parametrize("interleave", [False, True])
+def test_new_worker_receipt_survives_stale_observation(context, monkeypatch, operation, interleave):
+    pf, ticket, client = context
+    revision = ticket.model_dump(mode="json")["updated_at"]
+    original_update = pf.update_ticket
+    evidence = {"process_executions": [{"process_id": "publish", "receipt_id": "failed"}],
+                "publication_authorization": {"status": "consumed_failed"}}
+
+    def worker_update():
+        original_update(
+            ticket.id, status="failed", execution=TicketExecution(state="failed", attempt=1),
+            outputs=TicketOutputs(result=evidence), actor="bot:worker", reason="apply_failed",
+        )
+
+    if interleave:
+        # The worker wins after the API/high-level pre-read but before the
+        # store mutation. A pre-read comparison alone cannot pass this case.
+        def race(ticket_id, **updates):
+            worker_update()
+            return original_update(ticket_id, **updates)
+        monkeypatch.setattr(pf, "update_ticket", race)
+    else:
+        worker_update()
+
+    async def unexpected_event(*_args):
+        pytest.fail("a rejected write must not broadcast a successful change")
+    monkeypatch.setattr(server, "_broadcast_ticket_event", unexpected_event)
+    body = {"expected_updated_at": revision, "actor": "bot:observer", "reason": "observation"}
+    body.update({"execution": {"state": "ready"}, "outputs": {"result": {"observed": True}}}
+                if operation == "update" else {"result": {"observed": True}})
+    response = client.post(f"/tickets/{ticket.id}/{operation}-if-current", json=body)
+    assert response.status_code == 409
+    assert response.json() == {"detail": "ticket_updated_at_precondition_failed"}
+    current = pf.get_ticket(ticket.id)
+    assert current.status == "failed"
+    assert current.execution.state == "failed"
+    assert current.execution.attempt == 1
+    assert current.outputs.result == evidence
+    assert current.history[-1]["actor"] == "bot:worker"
+
+
+def test_conditional_completion_keeps_governed_receipt_gate(context):
+    pf, ticket, client = context
+    ticket = pf.update_ticket(ticket.id, labels=["process-envelope:v2"])
+    response = client.post(
+        f"/tickets/{ticket.id}/complete-if-current",
+        json={"expected_updated_at": ticket.model_dump(mode="json")["updated_at"]},
+    )
+    assert response.status_code == 409
+    assert response.json() == {"detail": "completion_receipt_required"}
+    assert pf.get_ticket(ticket.id).status == "open"