Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 25 additions & 3 deletions planfile/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -311,9 +311,22 @@ def next_ticket(self, sprint: str = "current", queue: str | None = None) -> Tick
)
return min(runnable, key=self._ticket_sort_key, default=None)

def update_ticket(self, ticket_id: str, reason: str | None = None, actor: str | None = None, **updates):
def update_ticket(
self,
ticket_id: str,
reason: str | None = None,
actor: str | None = None,
expected_updated_at: str | None = None,
**updates,
):
"""Delegate with optional reason (why status/etc changed) and actor (who/by)."""
return self.store.update_ticket(ticket_id, reason=reason, actor=actor, **updates)
return self.store.update_ticket(
ticket_id,
reason=reason,
actor=actor,
expected_updated_at=expected_updated_at,
**updates,
)

def append_ticket_evidence(
self,
Expand Down Expand Up @@ -427,7 +440,15 @@ def complete_ticket(
execution = TicketExecution(**execution_data)
return self.update_ticket(ticket_id, status="done", execution=execution, outputs=outputs, reason=reason, actor=actor)

def fail_ticket(self, ticket_id: str, error: str, *, reason: str | None = None, actor: str | None = None) -> Ticket | None:
def fail_ticket(
self,
ticket_id: str,
error: str,
*,
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:
return None
Expand All @@ -452,6 +473,7 @@ def fail_ticket(self, ticket_id: str, error: str, *, reason: str | None = None,
execution=execution,
reason=reason or error,
actor=actor,
expected_updated_at=expected_updated_at,
)

def block_ticket(self, ticket_id: str, reason: str | None = None, note: str | None = None, *, actor: str | None = None) -> Ticket | None:
Expand Down
42 changes: 38 additions & 4 deletions planfile/api/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@
TicketOutputs,
TicketSource,
)
from planfile.core.store import ImmutableTerminalReopenError
from planfile.core.store import ImmutableTerminalReopenError, TicketUpdatedAtConflictError
from planfile.runtime_context import (
DEFAULT_CONFIG as DEFAULT_RUNTIME_CONFIG,
)
Expand Down Expand Up @@ -73,6 +73,8 @@ async def lifespan(_: FastAPI):
lifespan=lifespan,
)

API_CAPABILITIES = ["ticket.fail.expected_updated_at"]


@app.exception_handler(ImmutableTerminalReopenError)
async def immutable_terminal_reopen_handler(
Expand All @@ -81,6 +83,17 @@ async def immutable_terminal_reopen_handler(
):
return JSONResponse(status_code=409, content={"detail": "immutable_terminal_reopen"})


@app.exception_handler(TicketUpdatedAtConflictError)
async def ticket_updated_at_conflict_handler(
_: Request,
__: TicketUpdatedAtConflictError,
):
return JSONResponse(
status_code=409,
content={"detail": "ticket_updated_at_precondition_failed"},
)

_cors_origins = [
origin.strip()
for origin in os.environ.get("PLANFILE_CORS_ORIGINS", "").split(",")
Expand Down Expand Up @@ -190,6 +203,11 @@ class TicketFailRequest(BaseModel):
error: str
reason: str | None = None
actor: str | None = None
expected_updated_at: str | None = None


class TicketFailIfCurrentRequest(TicketFailRequest):
expected_updated_at: str


class TicketInputRequest(BaseModel):
Expand Down Expand Up @@ -825,8 +843,7 @@ async def complete_ticket(ticket_id: str, body: TicketCompleteRequest):
return ticket.model_dump(mode="json", exclude_none=True)


@app.post("/tickets/{ticket_id}/fail", tags=["tickets"])
async def fail_ticket(ticket_id: str, body: TicketFailRequest):
async def _fail_ticket(ticket_id: str, body: TicketFailRequest):
pf = get_planfile()
current = pf.get_ticket(ticket_id)
if not current:
Expand All @@ -837,13 +854,26 @@ async def fail_ticket(ticket_id: str, body: TicketFailRequest):
error=body.error,
reason=body.reason or body.error,
actor=body.actor or "unknown:api",
expected_updated_at=body.expected_updated_at,
)
if not ticket:
raise HTTPException(404, f"Ticket {ticket_id} not found")
await _broadcast_ticket_event("ticket.execution.changed", "fail", ticket)
return ticket.model_dump(mode="json", exclude_none=True)


@app.post("/tickets/{ticket_id}/fail", tags=["tickets"])
async def fail_ticket(ticket_id: str, body: TicketFailRequest):
return await _fail_ticket(ticket_id, body)


@app.post("/tickets/{ticket_id}/fail-if-current", tags=["tickets"])
async def fail_ticket_if_current(ticket_id: str, body: TicketFailIfCurrentRequest):
"""Fail a ticket only when it is still the exact observed revision."""

return await _fail_ticket(ticket_id, body)


@app.post("/tickets/{ticket_id}/input-required", tags=["tickets"])
async def wait_for_input(ticket_id: str, body: TicketInputRequest):
pf = get_planfile()
Expand Down Expand Up @@ -3177,7 +3207,11 @@ async def websocket_dsl(websocket: WebSocket, project_path: str = "."):
@app.get("/health", tags=["system"])
def health():
import planfile
return {"status": "ok", "version": planfile.__version__}
return {
"status": "ok",
"version": planfile.__version__,
"capabilities": API_CAPABILITIES,
}


@app.get("/", response_class=HTMLResponse, tags=["system"])
Expand Down
4 changes: 3 additions & 1 deletion planfile/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -158,8 +158,9 @@ def fail(
error: str,
reason: str | None = None,
actor: str | None = None,
expected_updated_at: str | None = None,
) -> TicketTransitionResult:
"""Record one failed execution attempt without deciding its retry policy."""
"""Record one failed attempt, optionally only for an observed ticket revision."""

return self._apply(
"fail",
Expand All @@ -168,6 +169,7 @@ def fail(
error=error,
reason=reason,
actor=actor,
expected_updated_at=expected_updated_at,
),
)

Expand Down
66 changes: 63 additions & 3 deletions planfile/core/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,10 @@ class ImmutableTerminalReopenError(RuntimeError):
"""Raised when an ordinary mutation tries to reactivate done/canceled work."""


class TicketUpdatedAtConflictError(RuntimeError):
"""Raised when a ticket changed after the caller observed it."""


class Store(StoreFileMixin, TicketStoreMixin):
"""File-based ticket store using .planfile/ directory."""

Expand Down Expand Up @@ -1833,14 +1837,61 @@ def _build_history_entry(
return entry

def update_ticket(
self, ticket_id: str, reason: str | None = None, actor: str | None = None, **updates
self,
ticket_id: str,
reason: str | None = None,
actor: str | None = None,
expected_updated_at: str | None = None,
**updates,
) -> Ticket | None:
"""Update a ticket. If status (or execution state) changes, a structured history entry
is appended automatically, including optional `reason` (why) and `actor` (who / by whom).
Use reason/actor (or _reason/_actor in **updates) for rich audit on status transitions.
"""
with self.mutation_lock():
return self._update_ticket_unlocked(ticket_id, reason=reason, actor=actor, **updates)
return self._update_ticket_unlocked(
ticket_id,
reason=reason,
actor=actor,
expected_updated_at=expected_updated_at,
**updates,
)

@staticmethod
def _updated_at_instant(value: object) -> datetime | None:
if isinstance(value, datetime):
parsed = value
elif isinstance(value, str) and value:
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return None
else:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=UTC)
return parsed.astimezone(UTC)

def _guard_expected_updated_at(
self,
previous: dict,
expected_updated_at: str | None,
) -> None:
if expected_updated_at is None:
return
projected = self._project_ticket_evidence(previous)
actual_updated_at = projected.get("updated_at")
actual_instant = self._updated_at_instant(actual_updated_at)
expected_instant = self._updated_at_instant(expected_updated_at)
if (
actual_instant is not None
and expected_instant is not None
and actual_instant == expected_instant
):
return
if str(actual_updated_at or "") == str(expected_updated_at):
return
raise TicketUpdatedAtConflictError("ticket_updated_at_precondition_failed")

def _guard_immutable_terminal_reopen(
self,
Expand Down Expand Up @@ -1987,14 +2038,20 @@ def append_ticket_evidence(
return model, True

def _update_ticket_unlocked(
self, ticket_id: str, reason: str | None = None, actor: str | None = None, **updates
self,
ticket_id: str,
reason: str | None = None,
actor: str | None = None,
expected_updated_at: str | None = None,
**updates,
) -> Ticket | None:
index_was_current = self._begin_index_mutation()
if self._uses_sharded_storage():
return self._update_ticket_sharded_unlocked(
ticket_id,
reason=reason,
actor=actor,
expected_updated_at=expected_updated_at,
_index_was_current=index_was_current,
**updates,
)
Expand All @@ -2007,6 +2064,7 @@ def _update_ticket_unlocked(
tickets = sprint_data.get("tickets", {})
if ticket_id in tickets:
previous = dict(tickets[ticket_id])
self._guard_expected_updated_at(previous, expected_updated_at)
# Extract history metadata (reason=why the change, actor/by=who performed it)
# Support both named params (from high-level methods) and _-prefixed or bare in updates
history_reason = (
Expand Down Expand Up @@ -2074,6 +2132,7 @@ def _update_ticket_sharded_unlocked(
ticket_id: str,
reason: str | None = None,
actor: str | None = None,
expected_updated_at: str | None = None,
_index_was_current: bool = False,
**updates,
) -> Ticket | None:
Expand All @@ -2083,6 +2142,7 @@ def _update_ticket_sharded_unlocked(
return None
sprint, ticket_data = located
previous = dict(ticket_data)
self._guard_expected_updated_at(previous, expected_updated_at)
history_reason = reason or updates.pop("reason", None) or updates.pop("_reason", None)
history_actor = actor or updates.pop("actor", None) or updates.pop("_actor", None)
serialized_updates = {
Expand Down
75 changes: 75 additions & 0 deletions tests/test_ticket_api_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,80 @@ def test_governed_ticket_mutations_require_attributed_history(tmp_path, monkeypa
assert updated.json()["history"][-1]["reason"] == "Escalated priority after preflight."


def test_fail_api_returns_conflict_for_stale_updated_at_precondition(tmp_path, monkeypatch):
pf = Planfile(str(tmp_path))
ticket = pf.create_ticket(
name="Watchdog candidate",
execution=TicketExecution(state="running", assigned_to="bot:worker", max_attempts=2),
)
observed_updated_at = ticket.model_dump(mode="json")["updated_at"]
changed = pf.update_ticket(ticket.id, priority="high")
assert changed is not None
monkeypatch.setattr(server, "get_planfile", lambda: pf)
client = TestClient(server.app)

conflict = client.post(
f"/tickets/{ticket.id}/fail-if-current",
json={
"error": "stale_execution_timeout",
"expected_updated_at": observed_updated_at,
},
)

assert conflict.status_code == 409
assert conflict.json() == {"detail": "ticket_updated_at_precondition_failed"}
current = pf.get_ticket(ticket.id)
assert current is not None
assert current.priority == "high"
assert current.execution.state == "running"
assert current.execution.attempt == 0


def test_fail_api_accepts_current_updated_at_in_json_timestamp_form(tmp_path, monkeypatch):
pf = Planfile(str(tmp_path))
ticket = pf.create_ticket(
name="Current watchdog candidate",
execution=TicketExecution(state="running", assigned_to="bot:worker", max_attempts=2),
)
expected_updated_at = ticket.model_dump(mode="json")["updated_at"]
monkeypatch.setattr(server, "get_planfile", lambda: pf)
client = TestClient(server.app)

response = client.post(
f"/tickets/{ticket.id}/fail",
json={
"error": "stale_execution_timeout",
"expected_updated_at": expected_updated_at,
},
)

assert response.status_code == 200
assert response.json()["execution"]["state"] == "ready"
assert response.json()["execution"]["attempt"] == 1


def test_fail_if_current_api_requires_updated_at_precondition(tmp_path, monkeypatch):
pf = Planfile(str(tmp_path))
ticket = pf.create_ticket(
name="Fail-closed watchdog candidate",
execution=TicketExecution(state="running", assigned_to="bot:worker"),
)
monkeypatch.setattr(server, "get_planfile", lambda: pf)
client = TestClient(server.app)

response = client.post(
f"/tickets/{ticket.id}/fail-if-current",
json={"error": "stale_execution_timeout"},
)

assert response.status_code == 422
current = pf.get_ticket(ticket.id)
assert current is not None
assert current.status == "open"
assert current.execution is not None
assert current.execution.attempt == 0


def test_governed_ticket_creation_requires_structured_four_part_envelope(tmp_path, monkeypatch):
pf = Planfile(str(tmp_path))
monkeypatch.setattr(server, "get_planfile", lambda: pf)
Expand Down Expand Up @@ -618,6 +692,7 @@ def test_openapi_and_health_publish_same_version():
client = TestClient(server.app)

assert client.get("/openapi.json").json()["info"]["version"] == client.get("/health").json()["version"]
assert "ticket.fail.expected_updated_at" in client.get("/health").json()["capabilities"]


def test_ticket_list_pagination_headers(tmp_path, monkeypatch):
Expand Down
Loading
Loading