From ccde4f477ffb7993fc7219244aecb0686f32180b Mon Sep 17 00:00:00 2001 From: MAxim Kiktev Date: Thu, 29 Oct 2026 08:06:39 +0000 Subject: [PATCH] fix(core): fence task completion by execution generation A late executor result must not finish a newer run of the same task. Compare the observed durable run count atomically, negotiate server support, and use the precondition for all automatic completion paths. Verified real HTTP stale/current transitions, five real executor/restart scenarios, compatibility checks and SDK tours. --- CHANGELOG.md | 2 + README.md | 2 +- README.ru.md | 2 +- ROADMAP.md | 2 + apps/web/src/api/schema.gen.ts | 6 ++ crates/ag-agent/src/tasks.rs | 52 ++++++++++++++--- crates/ag-agent/tests/agents.rs | 19 ++++++- crates/ag-protocol/proto/agwp/v1/agwp.proto | 2 + crates/ag-protocol/src/lib.rs | 4 ++ crates/ag-server/src/api/agents.rs | 11 +++- crates/ag-server/tests/task_attempts.rs | 18 ++++++ dev/goal/backlog.md | 2 + dev/goal/evidence.md | 18 ++++++ docs/adr/0035-task-completion-generation.md | 24 ++++++++ docs/adr/README.md | 2 + docs/architecture/adr-traceability.md | 9 +++ docs/protocol/README.md | 2 + docs/protocol/openapi.json | 10 ++++ proto/agwp/v1/agwp.proto | 2 + sdk/python/src/agentgit/_generated/models.py | 1 + .../src/agentgit/_generated/namespaces.py | 6 +- sdk/python/tests/test_all_operations.py | 4 +- sdk/typescript/src/generated/schema.ts | 6 ++ sdk/typescript/test/all-operations.test.ts | 8 ++- .../agent-orchestrator/internal/agwp/api.go | 20 +++++-- .../internal/agwp/task_finish_test.go | 56 +++++++++++++++++++ .../internal/scheduler/scheduler.go | 13 +++-- .../internal/scheduler/scheduler_test.go | 32 +++++++++++ .../internal/session/manager.go | 6 +- .../internal/session/session.go | 6 +- 30 files changed, 311 insertions(+), 36 deletions(-) create mode 100644 docs/adr/0035-task-completion-generation.md create mode 100644 services/agent-orchestrator/internal/agwp/task_finish_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 08f1561c..c4bf9318 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,8 @@ on-disk formats are versioned and refuse to open when newer than the build under ### Fixed +- Automatic task completion is fenced by the observed execution generation. A delayed result from an older run cannot complete or cancel a newer attempt of the same task. + - Automatic task cancellation now compares the observed attempt policy in the server transaction. Changing a limit during execution cannot be overwritten by a stale cancellation decision. - Integration rechecks the proposal head and open state together with the stream inside its commit transaction. A proposal updated or closed after candidate computation aborts the old operation, including the already-integrated path. diff --git a/README.md b/README.md index 2f0e5260..3e4022d6 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,7 @@ This is an early public beta: [status and limits](#status-and-limits) are below, **Development (`main`):** operators can disable or re-enable new runs for an individual agent from the browser or `ag agent disable NAME --remote origin` (`enable` to restore admission). Existing sessions, keys and history are retained. This is not included in the beta.5 archives; see [ADR-0026](docs/adr/0026-agent-run-admission.md). -**Development core:** Tasks can carry a server-enforced attempt limit. `ag task attempts TASK --max 5 --expected 3` lets an operator extend an exhausted task without losing its history; restart recovery uses durable run records. Zero keeps the existing repository retry policy. Concurrent operator policy changes are protected from stale automatic cancellation. [Execution limits](docs/adr/0033-task-attempt-limits.md). +**Development core:** Tasks can carry a server-enforced attempt limit. `ag task attempts TASK --max 5 --expected 3` lets an operator extend an exhausted task without losing its history; restart recovery uses durable run records. Zero keeps the existing repository retry policy. Concurrent operator policy changes are protected from stale automatic cancellation, and late results cannot finish a newer task attempt. [Execution limits](docs/adr/0033-task-attempt-limits.md). **Development upgrade:** persistent follow-up task links and durable task execution limits require metadata schema 8. Back up before opening existing repositories with this development build; rollback uses the pre-upgrade backup. [Source-link semantics](docs/adr/0027-task-source-links.md). diff --git a/README.ru.md b/README.ru.md index cd68550a..dafb8685 100644 --- a/README.ru.md +++ b/README.ru.md @@ -37,7 +37,7 @@ AgentGit — система контроля версий для репозит **В разработке (`main`):** оператор может отключить или снова разрешить новые запуски отдельного агента из браузера или через `ag agent disable NAME --remote origin` (`enable` для включения). Ключ, история и работающие сессии сохраняются. В архивы beta.5 эта функция не включена; [ADR-0026](docs/adr/0026-agent-run-admission.md). -**Ядро в разработке:** Сервер ограничивает число запусков каждой задачи, сохраняя счётчик через историю запусков при перезапуске. Оператор может продолжить исчерпанную задачу командой `ag task attempts TASK --max 5 --expected 3`, не теряя её историю. Нулевой лимит сохраняет прежнюю политику повторов репозитория. Изменение лимита во время запуска защищено от отмены по устаревшей политике. [Лимиты исполнения](docs/adr/0033-task-attempt-limits.md). +**Ядро в разработке:** Сервер ограничивает число запусков каждой задачи, сохраняя счётчик через историю запусков при перезапуске. Оператор может продолжить исчерпанную задачу командой `ag task attempts TASK --max 5 --expected 3`, не теряя её историю. Нулевой лимит сохраняет прежнюю политику повторов репозитория. Изменение лимита во время запуска защищено от отмены по устаревшей политике; запоздавший результат не завершит более новую попытку. [Лимиты исполнения](docs/adr/0033-task-attempt-limits.md). **Обновление разработки:** связи задач-продолжений и постоянные лимиты исполнения используют схему метаданных 8. Перед запуском этой сборки на существующих данных сделайте резервную копию; для отката восстановите её. [Семантика связей](docs/adr/0027-task-source-links.md). diff --git a/ROADMAP.md b/ROADMAP.md index 3e5dec94..8ee24060 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -157,3 +157,5 @@ hashing, merge semantics or the publication protocol need an ADR ([CONTRIBUTING] Merged in [PR #116](https://github.com/Paffin/AgentGit/pull/116): server-transactional attempt admission, operator extension with optimistic concurrency, CLI/HTTP/gRPC/SDK contracts, and orchestrator restart recovery. Metadata schema 8 preserves existing task/run history. Evidence: agent coordination, real-server executor recovery, HTTP permissions/idempotency, CLI and SQLite/PostgreSQL upgrade scenarios. Web UI displays exhaustion; editing the limit in the browser remains follow-up work. Automatic cancellation also carries an expected attempt policy. Concurrent operator changes survive stale session decisions, and reconciliation reloads durable history (ADR0034). + +Task completion is also bound to its observed execution generation. A late executor or stale scheduler cannot finish a newer attempt; capability negotiation protects older-server deployments (ADR0035). diff --git a/apps/web/src/api/schema.gen.ts b/apps/web/src/api/schema.gen.ts index 5b625acd..f0b370b6 100644 --- a/apps/web/src/api/schema.gen.ts +++ b/apps/web/src/api/schema.gen.ts @@ -2541,6 +2541,12 @@ export interface components { * log as an operator's, with the owner it overrode. */ FinishTaskRequest: { + /** + * Format: int64 + * @description Reject a stale completion when another run was admitted after this task snapshot. + * @default null + */ + expected_attempts_started: number | null /** * Format: int32 * @description Optional execution-policy precondition, checked atomically with the state change. diff --git a/crates/ag-agent/src/tasks.rs b/crates/ag-agent/src/tasks.rs index 12b7c6aa..e14f79e4 100644 --- a/crates/ag-agent/src/tasks.rs +++ b/crates/ag-agent/src/tasks.rs @@ -257,18 +257,18 @@ pub(crate) fn claim_in_tx(tx: &Tx<'_>, task: &str, owner: &str) -> Result<(), Me /// (`agent:bot#run_…`) may only finish the task *that run* claimed - another run of the same agent has no say in /// it. Use [`finish_task_as_operator`] for an operator that overrides the owner. pub fn finish_task(meta: &Meta, task: &str, state: &str, actor: &str) -> Result<(), AgentError> { - finish_task_with_limit(meta, task, state, actor, false, None) + finish_task_checked(meta, task, state, actor, false, TaskFinishPreconditions::default()) } /// Like [`finish_task`] for an operator: the task is finished whoever owns it. The audit entry says that it was /// an operator's doing and whose task it overrode, so the log never shows an owner finishing what an operator /// finished for it. pub fn finish_task_as_operator(meta: &Meta, task: &str, state: &str, actor: &str) -> Result<(), AgentError> { - finish_task_with_limit(meta, task, state, actor, true, None) + finish_task_checked(meta, task, state, actor, true, TaskFinishPreconditions::default()) } -/// Finishes only if the optional execution-policy precondition still matches. -/// The check shares the task transition transaction; zero is a meaningful expected limit. +/// Finishes with an optional retry-policy comparison. Prefer the full preconditions +/// when completing work from a particular execution attempt. pub fn finish_task_with_limit( meta: &Meta, task: &str, @@ -276,6 +276,46 @@ pub fn finish_task_with_limit( actor: &str, operator: bool, expected_max_attempts: Option, +) -> Result<(), AgentError> { + finish_task_checked( + meta, + task, + state, + actor, + operator, + TaskFinishPreconditions { max_attempts: expected_max_attempts, attempts_started: None }, + ) +} + +/// Observed execution state for a conditional task transition. +#[derive(Clone, Copy, Debug, Default)] +pub struct TaskFinishPreconditions { + /// Expected configured limit, including an explicitly unset limit of zero. + pub max_attempts: Option, + /// Expected durable count of admitted runs; a new run invalidates this snapshot. + pub attempts_started: Option, +} + +impl TaskFinishPreconditions { + fn check(self, task: &TaskRow) -> Result<(), MetaError> { + if self.max_attempts.is_some_and(|expected| task.max_attempts != expected) + || self.attempts_started.is_some_and(|expected| task.attempts_started != expected) + { + return Err(MetaError::Invalid("task execution changed; reload before finishing".into())); + } + Ok(()) + } +} + +/// Finishes only if the observed execution generation and policy still match. +/// Preconditions share the task transition transaction; zero remains meaningful. +pub fn finish_task_checked( + meta: &Meta, + task: &str, + state: &str, + actor: &str, + operator: bool, + expected: TaskFinishPreconditions, ) -> Result<(), AgentError> { if !["done", "cancelled"].contains(&state) { return Err(AgentError::Invalid("state must be 'done' or 'cancelled'".into())); @@ -286,9 +326,7 @@ pub fn finish_task_with_limit( }; meta.write(|tx| { let t = tx.task(task)?.ok_or_else(|| MetaError::NotFound(format!("task '{task}'")))?; - if expected_max_attempts.is_some_and(|expected| t.max_attempts != expected) { - return Err(MetaError::Invalid("task attempt limit changed; reload before finishing".into())); - } + expected.check(&t)?; let mut details = json!({"state": state}); match t.owner.as_ref() { Some(o) if operator => { diff --git a/crates/ag-agent/tests/agents.rs b/crates/ag-agent/tests/agents.rs index bce682dc..34e413ec 100644 --- a/crates/ag-agent/tests/agents.rs +++ b/crates/ag-agent/tests/agents.rs @@ -827,10 +827,25 @@ fn stale_retry_policy_cannot_cancel_a_task_or_append_a_finish_audit() { let task = tasks::create_task(&e.meta, "Preserve operator policy", "", &[], "user:creator").unwrap(); e.meta.write(|tx| tasks::set_attempt_limit_in(tx, &task.id, 2, 0, "user:operator", true)).unwrap(); let before = e.meta.read(|tx| tx.audit_list(None, 100)).unwrap().len(); - let stale = tasks::finish_task_with_limit(&e.meta, &task.id, "cancelled", "user:operator", true, Some(0)); + let stale = tasks::finish_task_checked( + &e.meta, + &task.id, + "cancelled", + "user:operator", + true, + tasks::TaskFinishPreconditions { max_attempts: Some(0), ..Default::default() }, + ); assert!(matches!(stale, Err(AgentError::Conflict(_)))); assert_eq!(e.meta.read(|tx| tx.task(&task.id)).unwrap().unwrap().state, "open"); assert_eq!(e.meta.read(|tx| tx.audit_list(None, 100)).unwrap().len(), before); - tasks::finish_task_with_limit(&e.meta, &task.id, "cancelled", "user:operator", true, Some(2)).unwrap(); + tasks::finish_task_checked( + &e.meta, + &task.id, + "cancelled", + "user:operator", + true, + tasks::TaskFinishPreconditions { max_attempts: Some(2), ..Default::default() }, + ) + .unwrap(); assert_eq!(e.meta.read(|tx| tx.task(&task.id)).unwrap().unwrap().state, "cancelled"); } diff --git a/crates/ag-protocol/proto/agwp/v1/agwp.proto b/crates/ag-protocol/proto/agwp/v1/agwp.proto index 124b17bc..965b3c4c 100644 --- a/crates/ag-protocol/proto/agwp/v1/agwp.proto +++ b/crates/ag-protocol/proto/agwp/v1/agwp.proto @@ -1227,6 +1227,8 @@ message FinishTaskRequest { // Optional execution-policy precondition, checked atomically with the state change. // Explicit zero matches an unset limit; omission preserves ordinary manual completion. optional uint32 expected_max_attempts = 5; + // Reject a stale completion when another run was admitted after this task snapshot. + optional uint64 expected_attempts_started = 6; } message Lease { diff --git a/crates/ag-protocol/src/lib.rs b/crates/ag-protocol/src/lib.rs index 1a5355aa..d9ef3830 100644 --- a/crates/ag-protocol/src/lib.rs +++ b/crates/ag-protocol/src/lib.rs @@ -89,6 +89,9 @@ pub mod features { /// Durable task-local attempt limits and optimistic policy updates. pub const TASK_ATTEMPT_LIMITS: &str = "tasks.attempt_limits"; + /// Task completion compares both the execution generation and retry policy. + pub const TASK_FINISH_PRECONDITIONS: &str = "tasks.finish_preconditions"; + /// Every feature this implementation knows. pub const ALL: &[&str] = &[ EVENTS_REPLAY, @@ -113,6 +116,7 @@ pub mod features { TASK_LINEAGE, TASK_EDIT, TASK_ATTEMPT_LIMITS, + TASK_FINISH_PRECONDITIONS, RUN_OUTPUT, TASK_ASSIGNMENT, EXECUTOR_PRESENCE, diff --git a/crates/ag-server/src/api/agents.rs b/crates/ag-server/src/api/agents.rs index 10c5398b..c50be432 100644 --- a/crates/ag-server/src/api/agents.rs +++ b/crates/ag-server/src/api/agents.rs @@ -4,7 +4,9 @@ use ag_agent::leases::{acquire_with, extend, release}; use ag_agent::predict::predict; use ag_agent::provenance::{ProvenanceRecord, SCHEMA, attach_signed, verify}; use ag_agent::registry::{finish_run_as, heartbeat, reap_runs, register_agent_with_key, start_run_with}; -use ag_agent::tasks::{add_dependency, create_task_with, finish_task_with_limit, unfinished_dependencies}; +use ag_agent::tasks::{ + TaskFinishPreconditions, add_dependency, create_task_with, finish_task_checked, unfinished_dependencies, +}; use ag_crypto::{PublicKey, Signature}; use ag_protocol::v1; use ag_storage::meta::{MetaError, TaskRow, Tx}; @@ -761,13 +763,16 @@ impl App { let repo_name = require(&req.repo, "repo")?.to_string(); let operator = self.authorize(call, &repo_name, Cap::AgentOperate, None).await.is_ok(); self.with_repo_mut(call, &repo_name, Cap::TaskWrite, None, move |repo, actor| { - finish_task_with_limit( + finish_task_checked( repo.meta(), &req.id, &req.state, actor.as_str(), operator, - req.expected_max_attempts, + TaskFinishPreconditions { + max_attempts: req.expected_max_attempts, + attempts_started: req.expected_attempts_started, + }, )?; let t = repo.meta().read(|tx| tx.task(&req.id))?.ok_or_else(|| ApiError::not_found("task not found"))?; task_view(repo, &t) diff --git a/crates/ag-server/tests/task_attempts.rs b/crates/ag-server/tests/task_attempts.rs index 43b065fb..641a8712 100644 --- a/crates/ag-server/tests/task_attempts.rs +++ b/crates/ag-server/tests/task_attempts.rs @@ -63,5 +63,23 @@ async fn attempts_are_counted_once_and_an_operator_can_extend_the_same_task() { assert_eq!(second.status, 200, "{}", second.json); assert_ne!(second.json["run_id"], started.json["run_id"]); assert_eq!(creator.get(&format!("{base}/tasks/{task}")).await.json["attempts_started"], 2); + let stale_run_finish = admin + .post( + &format!("{base}/tasks/{task}/finish"), + json!({"state":"cancelled", "expected_max_attempts":2, "expected_attempts_started":1}), + ) + .await; + assert_eq!(stale_run_finish.status, 412, "{}", stale_run_finish.json); + assert_eq!(creator.get(&format!("{base}/tasks/{task}")).await.json["state"], "in_progress"); + let second_run = second.json["run_id"].as_str().unwrap(); + assert_eq!( + admin.post(&format!("{base}/runs/{second_run}/complete"), json!({"status":"completed"})).await.status, + 200 + ); + let current_finish = admin + .post(&format!("{base}/tasks/{task}/finish"), json!({"state":"done", "expected_attempts_started":2})) + .await; + assert_eq!(current_finish.status, 200, "{}", current_finish.json); + assert_eq!(current_finish.json["state"], "done"); server.stop().await; } diff --git a/dev/goal/backlog.md b/dev/goal/backlog.md index 8e75292f..898ca6ea 100644 --- a/dev/goal/backlog.md +++ b/dev/goal/backlog.md @@ -99,3 +99,5 @@ CLI explanations use the same server rule. The active priority remains agent-nat Core integration: PR112 (atomic expected-head proposal updates) and PR114 (revalidate proposal and stream at final integration commit) are merged. PR115 (policy-required thread resolution) is also merged. PR113 adds review suggestions on the same head-precondition contract. These remain distinct from a published release. Retry policy changes during an active run now reject stale automatic cancellation atomically; the scheduler rereads durable history before resuming. See ADR0034 and the evidence journal. + +Execution completion now carries a durable generation precondition across core, API and orchestrator (ADR0035). A late result cannot finish a newer attempt; matching manual completion remains available. diff --git a/dev/goal/evidence.md b/dev/goal/evidence.md index d9e35439..37925288 100644 --- a/dev/goal/evidence.md +++ b/dev/goal/evidence.md @@ -601,3 +601,21 @@ history (0.01s); HTTP stale cancellation returned 412 (0.14s). SDK tours passed `/workspace/agentgit-quality/task-policy-finish-red.log`; follow-up logs share that prefix. No schema migration; ADR0034 describes the additive protocol field. Static Crapper/Dryer/ Mutator scans were run; these are not coverage or mutation-kill measurements. + +## Completion is bound to the observed task execution + +ADR0035 adds `expected_attempts_started` to task completion. It is checked together with the +optional expected attempt limit in the same state/audit transaction. Automatic session, +recovered-budget and landed-proposal completion carry the observed generation. The +orchestrator negotiates `tasks.finish_preconditions` and leaves a task unchanged if an old +server would ignore the condition. Explicit manual completion remains compatible. + +Red: a real HTTP request carrying generation one cancelled a task already on its second +run (200 instead of 412). Green: stale generation is rejected and matching completion succeeds +(0.14s); previous policy/state/audit regression remains green. Five real-server/shell scheduler +scenarios passed in 1.50s, including a delayed first executor after a second run was admitted, +mid-run policy changes, restart and legacy cancellation. Protocol checks cover unsupported +servers and preservation of explicit zero (0.008s). TypeScript/Python SDK tours passed +(12/13 tests) using both conditions; Web types remain valid. Crapper/Dryer/Mutator scans are +static inspection, not a measured mutation score. Logs: task-completion-fence-* in the +implementation workspace. diff --git a/docs/adr/0035-task-completion-generation.md b/docs/adr/0035-task-completion-generation.md new file mode 100644 index 00000000..8e210c79 --- /dev/null +++ b/docs/adr/0035-task-completion-generation.md @@ -0,0 +1,24 @@ +# ADR 0035: fence automatic task completion by execution generation + +Status: accepted. + +## Context + +An orchestrator finishes tasks with operator credentials after its run credential stops +working. A delayed result from an earlier attempt must not finish work owned by a newer +attempt, even if both attempts use the same agent and retry limit. + +## Decision + +Add optional `expected_attempts_started` to `FinishTaskRequest` (protobuf field 6). The core +compares it to the durable count of admitted task runs in the same write transaction as +state and audit. A newer admitted run changes that count and stale completion is rejected +with HTTP 412 / failed_precondition, without changing the task or adding a finish audit. +The count includes failed and abandoned runs and is never reset by retry-policy edits. + +Automatic session completion, cancellation, history recovery and proposal-result completion +supply the observed count. Cancellation also supplies the expected retry limit (ADR0034). +On a rejected transition, the scheduler discards local assumptions and reloads the task and +its history. Explicit manual completion without preconditions keeps its existing behavior. +Servers advertise `tasks.finish_preconditions`; the orchestrator verifies it before automatic completion and leaves the task unchanged on an older server. No storage migration or canonical-format change is required. Upgrade all replicas and +orchestrators together; this extends the unreleased task execution contract. diff --git a/docs/adr/README.md b/docs/adr/README.md index 028b607e..78d9dc9b 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -44,3 +44,5 @@ which test: [ADR traceability](../architecture/adr-traceability.md). - [0033 — Task execution-attempt limits](0033-task-attempt-limits.md) - [0034 — Task completion policy precondition](0034-task-finish-policy-precondition.md) + +- [0035 — Task completion execution generation](0035-task-completion-generation.md) diff --git a/docs/architecture/adr-traceability.md b/docs/architecture/adr-traceability.md index fc1eae85..b93ddd57 100644 --- a/docs/architecture/adr-traceability.md +++ b/docs/architecture/adr-traceability.md @@ -114,3 +114,12 @@ NATS 🔜 (events use SSE and the audit log) · Docker Compose/Kubernetes 🟡 ( ## ADR 0015 — revision-bound content search The new read-only `tree.search` operation uses the existing `TreeRead` authorization, immutable tree and blob interfaces. Focused HTTP checks cover bounded continuation, skipped content, cursor mismatch and repository isolation; `code-search.spec.ts` exercises UI-created content, pagination and line navigation. See [ADR 0015](../adr/0015-bounded-revision-code-search.md). No index-latency or complete binary/large-file search claim is made. + +## ADR0033–0035 — durable task execution and completion preconditions + +`crates/ag-agent/tests/agents.rs` covers durable attempt admission, rollback and preservation +of state/audit after a stale policy update. `crates/ag-server/tests/task_attempts.rs` exercises +the wire contract, stale generation rejection and current-generation completion. The real +server/shell-executor scheduler scenarios cover restart, operator policy changes during work +and a late first session while a second run owns the task. Capability tests refuse an older +server before sending a conditional completion. diff --git a/docs/protocol/README.md b/docs/protocol/README.md index 23524054..84df997b 100644 --- a/docs/protocol/README.md +++ b/docs/protocol/README.md @@ -45,3 +45,5 @@ Clients: the `ag` CLI (`ag clone/fetch/pull/push/remote`) and the Rust crate `ag Negotiate `tasks.attempt_limits` before relying on enforcement. `Task.max_attempts` is the total allowed admitted runs (0 means unset); `Task.attempts_started` is derived from durable run history. Creation accepts `max_attempts`. `PATCH /v1/repos/{owner}/{rname}/tasks/{id}/attempt-limit` (`task.set_attempt_limit`, gRPC `SetTaskAttemptLimit`) requires `max_attempts`, `expected_max_attempts`, and an idempotency key. The creator can adjust the limit before the first run; subsequent changes require an operator. Exhausted tasks are not ready. A rejected start consumes no attempt, and an accepted start replay returns the original run. `task.finish` also accepts optional `expected_max_attempts`. The state transition compares this value atomically; a changed limit returns HTTP 412 without finishing the task. Explicit zero compares an unset policy. Omission retains manual completion semantics. + +Negotiate `tasks.finish_preconditions` before depending on either completion precondition. To complete a particular execution, pass `task.finish.expected_attempts_started` from the claimed task snapshot. A newer admitted run invalidates that precondition, preventing an earlier executor from finishing newer work. Automatic orchestrator completion always uses it. diff --git a/docs/protocol/openapi.json b/docs/protocol/openapi.json index f6bf01d1..922d4b90 100644 --- a/docs/protocol/openapi.json +++ b/docs/protocol/openapi.json @@ -1559,6 +1559,16 @@ "FinishTaskRequest": { "description": "Only the task's owner may finish it - or an operator (`agent.operate`), whose action is recorded in the audit\nlog as an operator's, with the owner it overrode.", "properties": { + "expected_attempts_started": { + "default": null, + "description": "Reject a stale completion when another run was admitted after this task snapshot.", + "format": "int64", + "minimum": 0, + "type": [ + "integer", + "null" + ] + }, "expected_max_attempts": { "default": null, "description": "Optional execution-policy precondition, checked atomically with the state change.\nExplicit zero matches an unset limit; omission preserves ordinary manual completion.", diff --git a/proto/agwp/v1/agwp.proto b/proto/agwp/v1/agwp.proto index 124b17bc..965b3c4c 100644 --- a/proto/agwp/v1/agwp.proto +++ b/proto/agwp/v1/agwp.proto @@ -1227,6 +1227,8 @@ message FinishTaskRequest { // Optional execution-policy precondition, checked atomically with the state change. // Explicit zero matches an unset limit; omission preserves ordinary manual completion. optional uint32 expected_max_attempts = 5; + // Reject a stale completion when another run was admitted after this task snapshot. + optional uint64 expected_attempts_started = 6; } message Lease { diff --git a/sdk/python/src/agentgit/_generated/models.py b/sdk/python/src/agentgit/_generated/models.py index b088c4e0..644a7519 100644 --- a/sdk/python/src/agentgit/_generated/models.py +++ b/sdk/python/src/agentgit/_generated/models.py @@ -758,6 +758,7 @@ class FileOp: @dataclass(kw_only=True) class FinishTaskRequest: """Only the task's owner may finish it - or an operator (`agent.operate`), whose action is recorded in the audit log as an operator's, with the owner it overrode.""" + expected_attempts_started: int | None = field(default=None) expected_max_attempts: int | None = field(default=None) id: str = field(default="") idempotency_key: str = field(default="") diff --git a/sdk/python/src/agentgit/_generated/namespaces.py b/sdk/python/src/agentgit/_generated/namespaces.py index f446bf95..415d5b06 100644 --- a/sdk/python/src/agentgit/_generated/namespaces.py +++ b/sdk/python/src/agentgit/_generated/namespaces.py @@ -1619,13 +1619,14 @@ def finish( *, repo: str, id: str, + expected_attempts_started: int | None = None, expected_max_attempts: int | None = None, idempotency_key: str | None = None, state: str | None = None, options: CallOptions | None = None, ) -> Task: """Mark a task done or cancelled (`POST /v1/repos/{owner}/{rname}/tasks/{id}/finish`, operation `task.finish`).""" - return self._c._call("task.finish", {"repo": repo, "id": id, "expected_max_attempts": expected_max_attempts, "idempotency_key": idempotency_key, "state": state}, options, Task) + return self._c._call("task.finish", {"repo": repo, "id": id, "expected_attempts_started": expected_attempts_started, "expected_max_attempts": expected_max_attempts, "idempotency_key": idempotency_key, "state": state}, options, Task) def get( self, @@ -3241,13 +3242,14 @@ async def finish( *, repo: str, id: str, + expected_attempts_started: int | None = None, expected_max_attempts: int | None = None, idempotency_key: str | None = None, state: str | None = None, options: CallOptions | None = None, ) -> Task: """Mark a task done or cancelled (`POST /v1/repos/{owner}/{rname}/tasks/{id}/finish`, operation `task.finish`).""" - return await self._c._call("task.finish", {"repo": repo, "id": id, "expected_max_attempts": expected_max_attempts, "idempotency_key": idempotency_key, "state": state}, options, Task) + return await self._c._call("task.finish", {"repo": repo, "id": id, "expected_attempts_started": expected_attempts_started, "expected_max_attempts": expected_max_attempts, "idempotency_key": idempotency_key, "state": state}, options, Task) async def get( self, diff --git a/sdk/python/tests/test_all_operations.py b/sdk/python/tests/test_all_operations.py index 0c5b0e5c..2fae40ca 100644 --- a/sdk/python/tests/test_all_operations.py +++ b/sdk/python/tests/test_all_operations.py @@ -351,7 +351,9 @@ def test_agents_tasks_leases_and_provenance(tour: dict[str, AgentGitClient]) -> ) assert entry.verified assert len(alice.provenance.list(repo=REPO, changeset=head).entries) == 1 - runner.tasks.finish(repo=REPO, id=task.id, state="done", expected_max_attempts=2) + runner.tasks.finish( + repo=REPO, id=task.id, state="done", expected_max_attempts=2, expected_attempts_started=1 + ) runner.agents.complete(repo=REPO, run_id=run.run_id, status="completed") assert alice.tasks.get(repo=REPO, id=dep.id).ready follow_up = alice.tasks.create(repo=REPO, title="Separate follow-up", source_task=task.id) diff --git a/sdk/typescript/src/generated/schema.ts b/sdk/typescript/src/generated/schema.ts index cf925ece..8fa63fc7 100644 --- a/sdk/typescript/src/generated/schema.ts +++ b/sdk/typescript/src/generated/schema.ts @@ -2556,6 +2556,12 @@ export interface components { * log as an operator's, with the owner it overrode. */ FinishTaskRequest: { + /** + * Format: int64 + * @description Reject a stale completion when another run was admitted after this task snapshot. + * @default null + */ + expected_attempts_started: number | null; /** * Format: int32 * @description Optional execution-policy precondition, checked atomically with the state change. diff --git a/sdk/typescript/test/all-operations.test.ts b/sdk/typescript/test/all-operations.test.ts index 2dfee172..d5f85982 100644 --- a/sdk/typescript/test/all-operations.test.ts +++ b/sdk/typescript/test/all-operations.test.ts @@ -359,7 +359,13 @@ describe('the tour', () => { const entry = await runner.provenance.submit({ repo, record, signature: signProvenance(record, key.privateKey) }); expect(entry.verified).toBe(true); expect((await alice.provenance.list({ repo, changeset: head })).entries).toHaveLength(1); - await runner.tasks.finish({ repo, id: task.id, state: 'done', expected_max_attempts: 2 }); + await runner.tasks.finish({ + repo, + id: task.id, + state: 'done', + expected_max_attempts: 2, + expected_attempts_started: 1, + }); await runner.agents.complete({ repo, run_id: run.run_id, status: 'completed' }); expect((await alice.tasks.get({ repo, id: dep.id })).ready).toBe(true); const followUp = await alice.tasks.create({ repo, title: 'Separate follow-up', source_task: task.id }); diff --git a/services/agent-orchestrator/internal/agwp/api.go b/services/agent-orchestrator/internal/agwp/api.go index bc12a95b..ea8b4ee8 100644 --- a/services/agent-orchestrator/internal/agwp/api.go +++ b/services/agent-orchestrator/internal/agwp/api.go @@ -186,16 +186,26 @@ func (c *Client) FinishTask(ctx context.Context, repo, id, state string) (*Task, return c.finishTask(ctx, repo, id, state, nil) } -// FinishTaskAtAttemptLimit rejects a decision based on an obsolete retry policy. -func (c *Client) FinishTaskAtAttemptLimit(ctx context.Context, repo, id, state string, expected uint32) (*Task, error) { - return c.finishTask(ctx, repo, id, state, &expected) +// FinishTaskAtSnapshot rejects a stale execution generation or cancellation policy. +func (c *Client) FinishTaskAtSnapshot(ctx context.Context, repo string, task Task, state string) (*Task, error) { + caps, err := c.Negotiate(ctx) + if err != nil { + return nil, err + } + if !caps.Has("tasks.finish_preconditions") { + return nil, fmt.Errorf("server lacks tasks.finish_preconditions; upgrade before automatic task completion") + } + return c.finishTask(ctx, repo, task.ID, state, &task) } -func (c *Client) finishTask(ctx context.Context, repo, id, state string, expected *uint32) (*Task, error) { +func (c *Client) finishTask(ctx context.Context, repo, id, state string, expected *Task) (*Task, error) { var out Task body := map[string]any{"state": state} if expected != nil { - body["expected_max_attempts"] = *expected + body["expected_attempts_started"] = expected.AttemptsStarted + if state == TaskCancelled { + body["expected_max_attempts"] = expected.MaxAttempts + } } if err := c.repoCall(ctx, "task.finish", http.MethodPost, repo, "/tasks/"+url.PathEscape(id)+"/finish", nil, body, &out); err != nil { return nil, err diff --git a/services/agent-orchestrator/internal/agwp/task_finish_test.go b/services/agent-orchestrator/internal/agwp/task_finish_test.go new file mode 100644 index 00000000..68614f35 --- /dev/null +++ b/services/agent-orchestrator/internal/agwp/task_finish_test.go @@ -0,0 +1,56 @@ +package agwp + +import ( + "context" + "encoding/json" + "net/http" + "testing" +) + +func TestConditionalFinishRequiresServerSupport(t *testing.T) { + writes := 0 + client, _, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/capabilities" { + _ = json.NewEncoder(w).Encode(Capabilities{ProtocolVersion: ProtocolVersion}) + return + } + writes++ + _ = json.NewEncoder(w).Encode(Task{}) + })) + _, err := client.FinishTaskAtSnapshot(context.Background(), "a/b", Task{ID: "task", AttemptsStarted: 1}, TaskDone) + if err == nil || writes != 0 { + t.Fatalf("unsupported server received completion: writes=%d err=%v", writes, err) + } +} + +func TestConditionalFinishPreservesZeroLimitAndExecutionGeneration(t *testing.T) { + for _, state := range []string{TaskDone, TaskCancelled} { + t.Run(state, func(t *testing.T) { + var sent map[string]any + client, _, _ := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/capabilities" { + _ = json.NewEncoder(w).Encode(Capabilities{ProtocolVersion: ProtocolVersion, Features: []string{"tasks.finish_preconditions"}}) + return + } + if err := json.NewDecoder(r.Body).Decode(&sent); err != nil { + t.Error(err) + } + _ = json.NewEncoder(w).Encode(Task{ID: "task", State: state}) + })) + _, err := client.FinishTaskAtSnapshot(context.Background(), "a/b", Task{ID: "task", AttemptsStarted: 3}, state) + if err != nil { + t.Fatal(err) + } + if sent["expected_attempts_started"] != float64(3) { + t.Fatalf("missing execution generation: %+v", sent) + } + limit, present := sent["expected_max_attempts"] + if state == TaskCancelled && (!present || limit != float64(0)) { + t.Fatalf("zero policy was omitted: %+v", sent) + } + if state == TaskDone && present { + t.Fatalf("successful execution was tied to a mutable retry policy: %+v", sent) + } + }) + } +} diff --git a/services/agent-orchestrator/internal/scheduler/scheduler.go b/services/agent-orchestrator/internal/scheduler/scheduler.go index 60114049..9014cbb8 100644 --- a/services/agent-orchestrator/internal/scheduler/scheduler.go +++ b/services/agent-orchestrator/internal/scheduler/scheduler.go @@ -47,8 +47,7 @@ type Sessions interface { FreeSlots(c config.Executor) int Cancel(runID, reason string) bool Has(runID string) bool - FinishTask(ctx context.Context, repo, taskID, state string) error - CancelTaskAtAttemptLimit(ctx context.Context, repo, taskID string, expected uint32) error + FinishTaskAtSnapshot(ctx context.Context, repo string, task agwp.Task, state string) error } // PathSource tells which files each open proposal changes (satisfied by @@ -465,7 +464,7 @@ func (s *Scheduler) recoverBudget(ctx context.Context, t agwp.Task, ex config.Ex } if spent >= s.o.Repo.Schedule.Attempts || temp >= maxTempFails { s.log.Warn("giving up on the task: its attempt budget was spent before this process started", "task", t.ID, "attempts", spent, "temporary_failures", temp) - if err := s.o.Sessions.CancelTaskAtAttemptLimit(ctx, repo, t.ID, t.MaxAttempts); err != nil && !agwp.IsCode(err, agwp.CodeFailedPrecond) { + if err := s.o.Sessions.FinishTaskAtSnapshot(ctx, repo, t, agwp.TaskCancelled); err != nil && !agwp.IsCode(err, agwp.CodeFailedPrecond) { s.log.Warn("could not cancel the task", "task", t.ID, "err", err) } s.clearDeferred(t.ID) @@ -762,8 +761,12 @@ func (s *Scheduler) resolveWaiting(ctx context.Context, tasks []agwp.Task, liveR default: state = agwp.TaskCancelled } - if err := s.o.Sessions.FinishTask(ctx, repo, id, state); err != nil && !agwp.IsCode(err, agwp.CodeFailedPrecond) { - s.log.Warn("could not finish the task", "task", id, "state", state, "err", err) + if err := s.o.Sessions.FinishTaskAtSnapshot(ctx, repo, t, state); err != nil { + if !agwp.IsCode(err, agwp.CodeFailedPrecond) { + s.log.Warn("could not finish the task", "task", id, "state", state, "err", err) + } else { + s.forgetTask(id) + } continue } s.log.Info("work landed or was closed; task finished", "task", id, "state", state) diff --git a/services/agent-orchestrator/internal/scheduler/scheduler_test.go b/services/agent-orchestrator/internal/scheduler/scheduler_test.go index ffa04b35..dd0798b3 100644 --- a/services/agent-orchestrator/internal/scheduler/scheduler_test.go +++ b/services/agent-orchestrator/internal/scheduler/scheduler_test.go @@ -890,3 +890,35 @@ func TestChangingAttemptPolicyDuringRunDoesNotCancelTheTask(t *testing.T) { t.Fatalf("unexpected attempts: %+v", h.task(task)) } } + +func TestOldSessionCannotCompleteANewerTaskAttempt(t *testing.T) { + release := filepath.Join(t.TempDir(), "release") + started := release + ".started" + ex := command(t, "stale-bot", fmt.Sprintf("touch '%[2]s'; while [ ! -f '%[1]s' ]; do sleep 0.01; done; exit 0", release, started)) + ex.Beat = time.Hour + h := newHarness(t, nil, nil, ex) + task := h.addTask("Keep the newer attempt", nil) + h.reconcile() + h.eventually("executor started", func() bool { _, err := os.Stat(started); return err == nil }) + first := h.runs()[0] + if _, err := h.w.Srv.Admin.CompleteRun(context.Background(), repo, first.RunID, "completed", "completed elsewhere"); err != nil { + t.Fatal(err) + } + newer, err := h.w.Srv.Admin.StartRun(context.Background(), repo, ex.Agent, task) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(release, nil, 0o600); err != nil { + t.Fatal(err) + } + h.waitIdle() + current := h.task(task) + if current.State != agwp.TaskInProgress || current.AttemptsStarted != 2 { + t.Fatalf("old session finished newer work: %+v", current) + } + for _, run := range h.runs() { + if run.RunID == newer.RunID && run.Status != "running" { + t.Fatalf("new run changed: %+v", run) + } + } +} diff --git a/services/agent-orchestrator/internal/session/manager.go b/services/agent-orchestrator/internal/session/manager.go index 5935bfd5..1c2bcd19 100644 --- a/services/agent-orchestrator/internal/session/manager.go +++ b/services/agent-orchestrator/internal/session/manager.go @@ -322,9 +322,9 @@ func (m *Manager) FinishTask(ctx context.Context, repo, taskID, state string) er return err } -// CancelTaskAtAttemptLimit preserves a task when its retry policy changed after scheduling. -func (m *Manager) CancelTaskAtAttemptLimit(ctx context.Context, repo, taskID string, expected uint32) error { - _, err := m.o.Orch.FinishTaskAtAttemptLimit(ctx, repo, taskID, "cancelled", expected) +// FinishTaskAtSnapshot prevents an earlier execution from finishing newer work. +func (m *Manager) FinishTaskAtSnapshot(ctx context.Context, repo string, task agwp.Task, state string) error { + _, err := m.o.Orch.FinishTaskAtSnapshot(ctx, repo, task, state) return err } diff --git a/services/agent-orchestrator/internal/session/session.go b/services/agent-orchestrator/internal/session/session.go index 06699585..3f05ef81 100644 --- a/services/agent-orchestrator/internal/session/session.go +++ b/services/agent-orchestrator/internal/session/session.go @@ -257,11 +257,7 @@ func (s *Session) finishTask(state string) { for attempt := 0; attempt < 3; attempt++ { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) // By the orchestrator, as an operator: the run's own credential has just stopped working with the run. - if state == "cancelled" { - _, err = s.m.o.Orch.FinishTaskAtAttemptLimit(ctx, s.req.Repo, s.req.Task.ID, state, s.req.Task.MaxAttempts) - } else { - _, err = s.m.o.Orch.FinishTask(ctx, s.req.Repo, s.req.Task.ID, state) - } + _, err = s.m.o.Orch.FinishTaskAtSnapshot(ctx, s.req.Repo, s.req.Task, state) cancel() if err == nil { return