From b6c6a959a631fbcfbeb968852778ad0d3aa81b82 Mon Sep 17 00:00:00 2001 From: Tom Softreck Date: Sat, 5 Sep 2026 16:04:13 +0200 Subject: [PATCH] fix(api): add atomic conditional update and completion endpoints --- planfile/__init__.py | 6 +- planfile/api/server.py | 27 +++++- tests/test_ticket_update_precondition.py | 104 +++++++++++++++++++++++ 3 files changed, 135 insertions(+), 2 deletions(-) create mode 100644 tests/test_ticket_update_precondition.py diff --git a/planfile/__init__.py b/planfile/__init__.py index 2f7dacd..08c23c9 100644 --- a/planfile/__init__.py +++ b/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, diff --git a/planfile/api/server.py b/planfile/api/server.py index ea5b42d..3f4db96 100644 --- a/planfile/api/server.py +++ b/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,6 +1154,7 @@ 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") @@ -1142,6 +1162,11 @@ async def complete_ticket(ticket_id: str, body: TicketCompleteRequest): 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) diff --git a/tests/test_ticket_update_precondition.py b/tests/test_ticket_update_precondition.py new file mode 100644 index 0000000..c3e2fff --- /dev/null +++ b/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"