Implement governed FAAS job lifecycle contract for Hermes#10
Implement governed FAAS job lifecycle contract for Hermes#10erikhinla wants to merge 17 commits into
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a governed worker execution model (FAAS) by updating schemas, database models, and API endpoints to support task claiming, lease management, and sequential reflections. A new worker-facing lifecycle API (worker_jobs.py) is added, and existing intake and skill routers are refactored to align with the new contract. The review feedback highlights several critical improvement opportunities in the new worker_jobs.py endpoints, specifically recommending a 409 Conflict when a different worker attempts to claim an active lease, and adding idempotency checks to the fail_job and escalate_job endpoints. Additionally, a potential race condition was flagged in the legacy write_reflection endpoint when concurrently generating sequence numbers.
| if job.status == JobStatus.ACTIVE.value and job.lease_expires_at and job.lease_expires_at > now: | ||
| return {"outcome": "already_running", "job": job_payload(job)} |
There was a problem hiding this comment.
When a job is already ACTIVE with an unexpired lease, any worker attempting to claim it receives a 200 OK response with {"outcome": "already_running"}.
To prevent other workers from mistakenly assuming they successfully claimed the job (or having to parse the 200 OK response body to detect a failed claim), the API should raise a 409 Conflict if a different worker tries to claim an active lease, while still returning already_running with 200 OK for idempotent retries by the same worker.
| if job.status == JobStatus.ACTIVE.value and job.lease_expires_at and job.lease_expires_at > now: | |
| return {"outcome": "already_running", "job": job_payload(job)} | |
| if job.status == JobStatus.ACTIVE.value and job.lease_expires_at and job.lease_expires_at > now: | |
| if job.claimed_by == request.worker_id: | |
| return {"outcome": "already_running", "job": job_payload(job)} | |
| raise HTTPException(status_code=409, detail="Task is currently leased by another worker") |
| if job.status != JobStatus.ACTIVE.value or job.claimed_by != request.worker_id: | ||
| raise HTTPException(status_code=409, detail="Only the worker holding the active claim may fail this task") |
There was a problem hiding this comment.
The fail_job endpoint currently lacks an idempotency check. If a worker retries a failed request (e.g., due to a network timeout), the second request will find the job in FAILED status (which is not ACTIVE) and raise a 409 Conflict error. Adding an idempotency check similar to complete_job ensures that retries are handled gracefully.
| if job.status != JobStatus.ACTIVE.value or job.claimed_by != request.worker_id: | |
| raise HTTPException(status_code=409, detail="Only the worker holding the active claim may fail this task") | |
| if job.status == JobStatus.FAILED.value and job.claimed_by == request.worker_id: | |
| return {"outcome": "already_recorded", "job": job_payload(job)} | |
| if job.status != JobStatus.ACTIVE.value or job.claimed_by != request.worker_id: | |
| raise HTTPException(status_code=409, detail="Only the worker holding the active claim may fail this task") |
| if job.status == JobStatus.ACTIVE.value and job.claimed_by != request.worker_id: | ||
| raise HTTPException(status_code=409, detail="Only the worker holding the active claim may escalate this task") |
There was a problem hiding this comment.
The escalate_job endpoint has two issues:
- It only enforces the worker claim check if the job is currently
ACTIVE. If the job is in another state (e.g.,COMPLETEDorQUEUED), anyone can escalate it, which bypasses state transition guards. - It lacks an idempotency check for retried escalation requests by the same worker.
Restricting escalation to ACTIVE jobs and adding an idempotency check for already ESCALATED jobs resolves both issues.
| if job.status == JobStatus.ACTIVE.value and job.claimed_by != request.worker_id: | |
| raise HTTPException(status_code=409, detail="Only the worker holding the active claim may escalate this task") | |
| if job.status == JobStatus.ESCALATED.value and job.claimed_by == request.worker_id: | |
| return {"outcome": "already_recorded", "job": job_payload(job)} | |
| if job.status != JobStatus.ACTIVE.value or job.claimed_by != request.worker_id: | |
| raise HTTPException(status_code=409, detail="Only the worker holding the active claim may escalate this task") |
| if reflection.sequence_number is None: | ||
| next_result = await db.execute( | ||
| select(func.coalesce(func.max(ReflectionRecord.sequence_number), 0) + 1) | ||
| .where(ReflectionRecord.task_id == reflection.task_id) | ||
| ) | ||
| sequence_number = next_result.scalar_one() |
There was a problem hiding this comment.
In the legacy compatibility write_reflection endpoint, if sequence_number is not provided, it is computed concurrently using max(sequence_number) + 1. Under concurrent requests for the same task_id, this will lead to a race condition where both requests compute the same sequence number, causing one to fail with a database unique constraint violation (uq_reflection_task_sequence).
While this is a legacy endpoint, it is recommended to either lock the job row (similar to the new /jobs/{task_id}/reflections endpoint) or handle the unique constraint violation gracefully.
1 participant
Purpose
This draft PR is stacked on #9 and implements the minimum Postgres-backed FAAS contract required before a Hermes worker adapter can execute a real proving task.
What this changes
openclaw,hermes,agent_zeroand risk tierslow,medium,high.artifact_productionas the bounded Hermes work type.task_idas the idempotency key and refusing duplicate queue entries.title,goal,inputs,output_required) with each job.GET /v1/jobs/{task_id}POST /v1/jobs/{task_id}/claimPOST /v1/jobs/{task_id}/completePOST /v1/jobs/{task_id}/failPOST /v1/jobs/{task_id}/escalatePOST /v1/jobs/{task_id}/reflections/v1/hermes/*endpoints with the existing API token dependency and makes reflection sequence aware.services/bizbrain_lite/migrations/002_faas_worker_claim_contract.sqlfor an existing Hetzner staging database.Authority and deployment boundary
Review gates before staging
job_recordsduplicates; the migration intentionally stops if idempotency cannot be enforced cleanly.Verification note
I inspected the branch diff and isolated the implementation from #9. I could not run local Python/tests in the current Codex session because the local command runner fails to launch a shell; runtime verification remains required on staging before this PR is promoted.