diff --git a/contracts/agents-api/index.md b/contracts/agents-api/index.md index 0c42e051b..ce1d4aa1d 100644 --- a/contracts/agents-api/index.md +++ b/contracts/agents-api/index.md @@ -88,7 +88,7 @@ Each item is Core's deliberate or native behavior where the official service beh - Input to a `none` Session is admitted synchronously; Core does not emulate the official asynchronous admission window. - An events request takes at most 64 events ([send input](./sessions-events.md#send-input)); the official limit is 16384. -- A function result that resumes a waiting Turn emits `turn.in_progress`, and cancelling a Turn that waits on a function result emits an interim `agent.session.in_progress`. +- A function result that resumes a waiting Turn emits `turn.in_progress`. - A function call the native Harness abandons, such as a Subagent call dropped when the root closes or interrupts the child or a MiniMax Code call past its 120 s timeout, is withdrawn: it stops being a required action and a later result for it returns 409 `conflict_error`. The official API has no withdrawal. - Attaching to the stream mid-Turn sends no catch-up Item snapshots. - The Items list includes in-progress and incomplete output Items, and keeps a failed function result's submitted `output`. diff --git a/contracts/agents-api/sessions-events.md b/contracts/agents-api/sessions-events.md index 1fbd4bf5b..5b43214e9 100644 --- a/contracts/agents-api/sessions-events.md +++ b/contracts/agents-api/sessions-events.md @@ -22,7 +22,7 @@ A Session's `status` and `last_active_at` derive from its latest root Turn and f | --- | --- | | `idle` | No Turn yet, or the latest Turn completed or was cancelled. Input reserved for a hosted Environment that is still provisioning also reads `idle` | | `in_progress` | The latest Turn is queued, running or waiting | -| `requires_action` | The latest Turn waits for a function result and no cancellation was requested, or input waits for a `self_hosted` machine to connect. `required_actions` lists `function_call` or `environment_connection` entries | +| `requires_action` | The latest Turn waits for a function result, including after a cancellation request until the Turn ends, or input waits for a `self_hosted` machine to connect. `required_actions` lists `function_call` or `environment_connection` entries | | `failed` | The latest Turn failed (`error` is "The execution could not complete."), input reserved for the Environment failed, initial input expired before admission, or the Environment failed to initialize (see [Environment initialization failure](#environment-initialization-failure)) | ```mermaid @@ -33,7 +33,7 @@ stateDiagram-v2 in_progress --> requires_action: awaits function result idle --> requires_action: awaits self_hosted machine requires_action --> in_progress: result or machine - requires_action --> idle: cancelled + requires_action --> idle: Turn ended in_progress --> failed: Turn failed idle --> failed: input expired or Environment failed failed --> in_progress: new input @@ -50,7 +50,7 @@ A Session stays usable after a Turn fails: new input starts a new Turn. Later re - **Empty batch.** `{"events": []}` checks that the Session exists and returns 202. It creates no Turn, Item or retry identity. - **Retries.** An `Idempotency-Key` of up to 128 bytes identifies the whole ordered batch. The same key and batch return 202 again without admitting anything twice; the same key with another batch returns 409 `idempotency_conflict`. A request without a key is always new. - **Messages.** On an idle Session a message batch starts a queued Turn. While a Turn runs, messages join it (steering); they never start a parallel Turn. Each message stays its own user Item, even when the Harness receives several as one prompt. -- **Cancellation.** A queued Turn is cancelled without a live Runtime. A running Turn is cancelled when the Runtime confirms it; completion can win that race. The Turn has stopped when it reads `cancelled`, not when the request returns. A cancellation on an idle Session with no pending input is accepted and has no effect; while an input reservation is pending, it returns 409. +- **Cancellation.** A queued Turn is cancelled without a live Runtime. A running Turn is cancelled when the Runtime confirms it; completion can win that race. The Turn has stopped when it reads `cancelled`, not when the request returns. A Turn that waits for a function result keeps the Session at `requires_action` with its `required_actions` until then, and the request emits no Session event. A cancellation on an idle Session with no pending input is accepted and has no effect; while an input reservation is pending, it returns 409. - **Function results.** `turn_id`, `call_id` and `success` are required; `output` and `error` are optional and nullable ([content rules](./message-content.md#function-results)). An identical repeated result returns 202 without another application or event. The result Item appears when the Harness applies the result; a result that cancellation prevents from being applied stays stored but produces no Item. - **Queueing.** A queued Turn starts when a Runtime that supports the Session's Harness and configuration is connected and one of Core's [`OAC_EXECUTION_CONCURRENCY`](../../docs/configuration.md#settings) work slots is free. A Session stays bound to the Runtime that first ran it. - **Execution availability.** While Core shuts down, or after its Worker loses execution ownership, requests return 503 `execution_unavailable`. @@ -121,7 +121,7 @@ A retry with the same `Idempotency-Key` and `stream: true` returns 201 with only - Codex command output streams as `agent.output.command_execution_output.delta` with the command's Item ID and output index. Native output quotas and text conversion apply, so the deltas are not a byte-exact capture; the completed Item is authoritative. - A reasoning Item follows one sequence: `item.added` in progress with an empty `summary`; for each `summary_text` part in order, `reasoning_summary_part.added` with empty text, `reasoning_summary_text.delta` events, then `reasoning_summary_text.done` and `reasoning_summary_part.done` once the next part starts or the Item ends; then `item.done`. `reasoning_summary_part.done` has `status` `incomplete` for a part a cancelled or failed Turn cut off, otherwise null. - A cancelled or failed Turn marks its unfinished Items `incomplete` and keeps their partial content. -- A function call stays in `required_actions` until the Harness applies its result, or cancellation or the Turn's end removes it. A repeated notification emits no new state. +- A function call stays in `required_actions` until the Harness applies its result or the Turn ends. A cancellation request does not remove it from a waiting Turn. A repeated notification emits no new state. ## Turns and Items diff --git a/contracts/agents-api/zh/index.md b/contracts/agents-api/zh/index.md index 816473583..57a4f2e98 100644 --- a/contracts/agents-api/zh/index.md +++ b/contracts/agents-api/zh/index.md @@ -1,7 +1,7 @@ --- title: "OpenAI 兼容性" source: contracts/agents-api/index.md -source_hash: d3d542a34c7cf38c798a2ec85373f9d554f83892f7c47b6773783b7dcf230c62 +source_hash: d65d6dc5d6a8ec195950a119b667964bf5bb651c469191235779d19a7220ff92 --- Core 实现下方固定版本的 OpenAI Agents API。本台账列出各项资源的覆盖情况、相对于官方服务的所有差异以及已知缺口。 @@ -91,7 +91,7 @@ Core 自身字段位于 `x_agents_core` 中（[Core extensions](../../../docs/zh - 提交给 `none` Session 的输入会同步完成准入；Core 不会模拟官方异步准入窗口。 - 一次 events 请求最多包含 64 个事件（[发送输入](sessions-events.md#send-input)）；官方上限为 16384。 -- 用于恢复等待中 Turn 的函数结果会发出 `turn.in_progress`；取消正在等待函数结果的 Turn 会发出临时 `agent.session.in_progress`。 +- 用于恢复等待中 Turn 的函数结果会发出 `turn.in_progress`。 - 原生 Harness 放弃的函数调用（例如根关闭或中断子级时被丢弃的 Subagent 调用，或超过 120 秒超时的 MiniMax Code 调用）会被撤回：它不再是待执行操作，之后为它提交结果会返回 409 `conflict_error`。官方 API 没有撤回机制。 - 在 Turn 中途接入流时，不会发送补发 Item 快照。 - Items 列表会包含进行中和未完成的输出 Items，并保留失败函数结果中已提交的 `output`。 diff --git a/contracts/agents-api/zh/sessions-events.md b/contracts/agents-api/zh/sessions-events.md index f44880d08..099ea2ed7 100644 --- a/contracts/agents-api/zh/sessions-events.md +++ b/contracts/agents-api/zh/sessions-events.md @@ -1,7 +1,7 @@ --- title: "Session、事件和历史" source: contracts/agents-api/sessions-events.md -source_hash: cc519d898ede4653a4e8abd0d59da6da9d88e2da1587870a75d5b1bab8809fe0 +source_hash: e8dfd2f21bdcf16832c60fb9fac2df42bc96553d7f1294dc2742c3eb455772a1 --- 本契约涵盖 Session 内部发生的事情：发送输入、实时事件流，以及读取 Turn、Item 和使用量的持久化历史。Session 资源本身（创建配置、重试标识、更新、列出和删除）见 [API 行为](wire-semantics.md)。消息和函数结果内容见[消息内容](message-content.md)。[Agents API 指南](../../../docs/zh/api/public-agent-api.md)展示了使用 SDK 和 HTTP 的调用方式。 @@ -24,7 +24,7 @@ Session 的 `status` 和 `last_active_at` 由其最新的根 Turn 以及仍在 | --- | --- | | `idle` | 尚无 Turn，或最新 Turn 已完成或已取消。为仍在预配的托管 Environment 预留的输入也显示为 `idle` | | `in_progress` | 最新 Turn 正在排队、运行或等待 | -| `requires_action` | 最新 Turn 正在等待函数结果且未请求取消，或输入正在等待 `self_hosted` 机器连接。`required_actions` 列出 `function_call` 或 `environment_connection` 条目 | +| `requires_action` | 最新 Turn 正在等待函数结果（请求取消后在 Turn 结束前也保持如此），或输入正在等待 `self_hosted` 机器连接。`required_actions` 列出 `function_call` 或 `environment_connection` 条目 | | `failed` | 最新 Turn 失败（`error` 为 "The execution could not complete."）、为 Environment 预留的输入失败、初始输入在接纳前到期，或 Environment 初始化失败（见 [Environment 初始化失败](#environment-initialization-failure)） | ```mermaid @@ -35,7 +35,7 @@ stateDiagram-v2 in_progress --> requires_action: awaits function result idle --> requires_action: awaits self_hosted machine requires_action --> in_progress: result or machine - requires_action --> idle: cancelled + requires_action --> idle: Turn ended in_progress --> failed: Turn failed idle --> failed: input expired or Environment failed failed --> in_progress: new input @@ -52,7 +52,7 @@ Turn 失败后 Session 仍可使用：新输入会启动一个新 Turn。后来 - **空批次。** `{"events": []}` 会检查 Session 是否存在并返回 202。它不会创建 Turn、Item 或重试标识。 - **重试。** 最长 128 字节的 `Idempotency-Key` 标识整个有序批次。相同密钥和批次会再次返回 202，而不会重复接纳任何内容；相同密钥搭配另一个批次会返回 409 `idempotency_conflict`。不带密钥的请求始终是新请求。 - **消息。** 在空闲 Session 上，消息批次会启动一个排队的 Turn。Turn 运行时，消息会加入其中（引导）；它们绝不会启动并行 Turn。每条消息都保留为独立的用户 Item，即使 Harness 将多条消息作为同一个提示接收。 -- **取消。** 排队的 Turn 无需活动 Runtime 即可取消。正在运行的 Turn 只有在 Runtime 确认后才会取消；完成操作可能赢得该竞争。读取到 `cancelled` 时才表示该 Turn 已停止，而不是请求返回时。在没有待处理输入的情况下，对空闲 Session 执行的取消会被接受且不产生任何效果；而在输入预留待处理期间，取消会返回 409。 +- **取消。** 排队的 Turn 无需活动 Runtime 即可取消。正在运行的 Turn 只有在 Runtime 确认后才会取消；完成操作可能赢得该竞争。读取到 `cancelled` 时才表示该 Turn 已停止，而不是请求返回时。正在等待函数结果的 Turn 在此之前会让 Session 保持 `requires_action` 及其 `required_actions`，且该请求不会发出 Session 事件。在没有待处理输入的情况下，对空闲 Session 执行的取消会被接受且不产生任何效果；而在输入预留待处理期间，取消会返回 409。 - **函数结果。** `turn_id`、`call_id` 和 `success` 为必填项；`output` 和 `error` 为可选项且可为空（[内容规则](message-content.md#function-results)）。重复提交完全相同的结果会返回 202，不会再次应用或发出事件。Harness 应用结果时才会出现结果 Item；如果取消操作导致结果无法应用，结果仍会存储，但不会产生 Item。 - **排队。** 当一个已连接且支持该 Session Harness 和配置的 Runtime 接入，并且 Core 的 [`OAC_EXECUTION_CONCURRENCY`](../../../docs/zh/configuration.md#settings) 工作槽位有一个空闲时，排队的 Turn 才会启动。Session 始终绑定到首次运行它的 Runtime。 - **执行可用性。** Core 关闭期间，或其 Worker 失去执行所有权后，请求返回 503 `execution_unavailable`。 @@ -123,7 +123,7 @@ Turn 失败后 Session 仍可使用：新输入会启动一个新 Turn。后来 - Codex 命令输出以 `agent.output.command_execution_output.delta` 流式传输，并携带命令的 Item ID 和输出索引。原生输出配额和文本转换规则适用，因此这些增量并非逐字节捕获的原始结果；完整 Item 才是权威结果。 - reasoning Item 遵循以下单一顺序：先发送 `summary` 为空的进行中 `item.added`；然后按顺序对每个 `summary_text` 部分发送文本为空的 `reasoning_summary_part.added` 和 `reasoning_summary_text.delta` 事件，并在下一个部分开始或 Item 结束时发送 `reasoning_summary_text.done` 和 `reasoning_summary_part.done`；最后发送 `item.done`。被取消或失败的 Turn 截断的部分，其 `reasoning_summary_part.done` 的 `status` 为 `incomplete`，否则为 null。 - 已取消或失败的 Turn 会将其未完成的 Item 标记为 `incomplete`，并保留其部分内容。 -- 函数调用会一直保留在 `required_actions` 中，直到 Harness 应用其结果，或取消操作或 Turn 结束将其移除。重复通知不会发出新状态。 +- 函数调用会一直保留在 `required_actions` 中，直到 Harness 应用其结果，或 Turn 结束将其移除；对等待中 Turn 的取消请求不会将其移除。重复通知不会发出新状态。 ## Turn 与 Item {#turns-and-items} diff --git a/services/core/internal/api/function_state_test.go b/services/core/internal/api/function_state_test.go index 4225b8d12..afabfb87a 100644 --- a/services/core/internal/api/function_state_test.go +++ b/services/core/internal/api/function_state_test.go @@ -46,3 +46,20 @@ func TestFunctionStateEventsUseTheirOwnSnapshot(t *testing.T) { } } } + +func TestWaitingTurnReadsRequiresActionAfterCancellationRequest(t *testing.T) { + session := sessions.Session{ID: "session", CreatedAt: time.Now(), Metadata: map[string]string{}, + Configuration: json.RawMessage(`{"agent":{"id":"agent_test","model":"model","tools":[]},"environment":{"type":"none"}}`), + RequiredActions: []v1.FunctionCallAction{{Type: "function_call", CallID: "call", Name: "lookup", TurnID: "turn"}}, + LastTurn: &sessions.Turn{ID: "turn", Status: sessions.TurnWaiting, CancelRequestedAt: time.Now()}, + } + value, err := sessionResponse(session, "") + if err != nil || value.Status != "requires_action" || len(value.RequiredActions) != 1 { + t.Fatal(value, err) + } + session.LastTurn.Status = sessions.TurnCancelled + session.RequiredActions = nil + if value, err = sessionResponse(session, ""); err != nil || value.Status != "idle" { + t.Fatal(value, err) + } +} diff --git a/services/core/internal/api/session_response.go b/services/core/internal/api/session_response.go index 5f244d58f..271545cb8 100644 --- a/services/core/internal/api/session_response.go +++ b/services/core/internal/api/session_response.go @@ -53,7 +53,7 @@ func sessionResponse(session sessions.Session, executorURL string) (v1.Session, switch turn.Status { case sessions.TurnQueued, sessions.TurnInProgress, sessions.TurnWaiting: response.Status = "in_progress" - if turn.CancelRequestedAt.IsZero() && len(session.RequiredActions) > 0 { + if len(session.RequiredActions) > 0 { response.Status = "requires_action" for _, action := range session.RequiredActions { response.RequiredActions = append(response.RequiredActions, v1.RequiredAction{ diff --git a/services/core/internal/db/queries/functions.sql b/services/core/internal/db/queries/functions.sql index e9b6298bb..ac3629b0e 100644 --- a/services/core/internal/db/queries/functions.sql +++ b/services/core/internal/db/queries/functions.sql @@ -15,7 +15,7 @@ WHERE s.tenant_id = $1 AND f.session_id = $2 AND f.turn_id = $3 AND f.call_id = -- name: ListPendingFunctionCalls :many SELECT f.* FROM function_calls f JOIN turns t ON t.session_id = f.session_id AND t.id = f.turn_id WHERE f.session_id = $1 AND f.turn_id = $2 AND NOT f.applied AND NOT f.withdrawn - AND t.status IN ('in_progress', 'waiting') AND t.cancel_requested_at IS NULL + AND t.status IN ('in_progress', 'waiting') ORDER BY f.created_at, f.call_id; -- name: SessionHasFunctionCall :one diff --git a/services/core/internal/db/sqlc/functions.sql.go b/services/core/internal/db/sqlc/functions.sql.go index fcb3aeb2b..181d2ccd7 100644 --- a/services/core/internal/db/sqlc/functions.sql.go +++ b/services/core/internal/db/sqlc/functions.sql.go @@ -132,7 +132,7 @@ func (q *Queries) GetFunctionCall(ctx context.Context, arg GetFunctionCallParams const listPendingFunctionCalls = `-- name: ListPendingFunctionCalls :many SELECT f.session_id, f.turn_id, f.call_id, f.executor_call_id, f.name, f.arguments, f.result, f.applied, f.created_at, f.withdrawn FROM function_calls f JOIN turns t ON t.session_id = f.session_id AND t.id = f.turn_id WHERE f.session_id = $1 AND f.turn_id = $2 AND NOT f.applied AND NOT f.withdrawn - AND t.status IN ('in_progress', 'waiting') AND t.cancel_requested_at IS NULL + AND t.status IN ('in_progress', 'waiting') ORDER BY f.created_at, f.call_id ` diff --git a/services/core/internal/persistence/postgres/sessionpg/functions_test.go b/services/core/internal/persistence/postgres/sessionpg/functions_test.go index 66c2aef1a..1a5c1144f 100644 --- a/services/core/internal/persistence/postgres/sessionpg/functions_test.go +++ b/services/core/internal/persistence/postgres/sessionpg/functions_test.go @@ -206,9 +206,12 @@ func TestLoadRequiredActionsReadsPendingCalls(t *testing.T) { t.Fatal(string(encoded), err) } exec(t, pool, `UPDATE turns SET cancel_requested_at = clock_timestamp() WHERE id = $1`, f.turn) - // The Turn as the caller read it still waits; the read of its calls sees - // the requested cancellation. - if actions, err := LoadRequiredActions(t.Context(), q, f.session, TurnFromRow(row)); err != nil || actions == nil || len(actions) != 0 { + // A waiting Turn keeps its actions until it ends, also after a cancellation request. + row, err = q.GetTurn(t.Context(), sqlc.GetTurnParams{TenantID: f.tenant, SessionID: f.session, ID: f.turn}) + if err != nil { + t.Fatal(err) + } + if actions, err := LoadRequiredActions(t.Context(), q, f.session, TurnFromRow(row)); err != nil || len(actions) != 2 { t.Fatalf("cancelling Turn actions %+v %v", actions, err) } } diff --git a/services/core/internal/sessions/cancellation.go b/services/core/internal/sessions/cancellation.go index 7523e61ae..30b295f3e 100644 --- a/services/core/internal/sessions/cancellation.go +++ b/services/core/internal/sessions/cancellation.go @@ -2,34 +2,6 @@ package sessions import "context" -// turnCancellation is what a cancellation request settles beyond recording the -// request, decided from the Turn as it was before the request. -type turnCancellation int - -const ( - // cancellationRequested only records the request: a running Turn ends when - // its execution stops, and a repeated request adds nothing. - cancellationRequested turnCancellation = iota - // cancellationEndsTurn ends a queued Turn, which never started, as - // cancelled. - cancellationEndsTurn - // cancellationReportsActivity reports the Session activity of a waiting - // Turn on its first request: it no longer waits on actions. - cancellationReportsActivity -) - -// decideCancellation decides what requesting the cancellation of turn, read -// before the request, settles. -func decideCancellation(turn Turn) turnCancellation { - switch { - case turn.Status == TurnQueued: - return cancellationEndsTurn - case turn.Status == TurnWaiting && turn.CancelRequestedAt.IsZero(): - return cancellationReportsActivity - } - return cancellationRequested -} - // TurnCancellationTx is the Session transaction CancelTurn runs in. type TurnCancellationTx interface { JournalTx @@ -46,37 +18,25 @@ type TurnCancellationTx interface { } // CancelTurn requests the cancellation of turn, the Session's active Turn read -// under the Session lock, and publishes what the request settles: a queued Turn -// ends cancelled, and a waiting Turn reports its Session activity. Both read -// the Turn again after the request, which set its timestamps. +// under the Session lock. A queued Turn never started and ends cancelled, read +// again after the request, which set its timestamps. Any other Turn ends when +// its execution stops, and a waiting Turn keeps its Session activity until then. func CancelTurn(ctx context.Context, tx TurnCancellationTx, turn Turn) error { - cancellation := decideCancellation(turn) if err := tx.RequestTurnCancel(ctx, turn.ID); err != nil { return err } - switch cancellation { - case cancellationEndsTurn: - cancelled, err := tx.LoadTurn(ctx, turn.ID) - if err != nil { - return err - } - ending, err := tx.LoadEnding(ctx, turn.ID) - if err != nil { - return err - } - return tx.ApplyTurnEnd(ctx, turn.ID, EndTurn(cancelled, ending)) - case cancellationReportsActivity: - cancelling, err := tx.LoadTurn(ctx, turn.ID) - if err != nil { - return err - } - usage, err := tx.LoadUsage(ctx) - if err != nil { - return err - } - return tx.AppendChanges(ctx, ActivityChange(cancelling, usage, nil)) + if turn.Status != TurnQueued { + return nil + } + cancelled, err := tx.LoadTurn(ctx, turn.ID) + if err != nil { + return err + } + ending, err := tx.LoadEnding(ctx, turn.ID) + if err != nil { + return err } - return nil + return tx.ApplyTurnEnd(ctx, turn.ID, EndTurn(cancelled, ending)) } // CancellationTx is the Session transaction CancelWork runs in. diff --git a/services/core/internal/sessions/cancellation_test.go b/services/core/internal/sessions/cancellation_test.go index 624b83202..3e71abc34 100644 --- a/services/core/internal/sessions/cancellation_test.go +++ b/services/core/internal/sessions/cancellation_test.go @@ -5,29 +5,10 @@ import ( "errors" "reflect" "testing" - "time" ) -func TestDecideCancellation(t *testing.T) { - requested := turnWith(TurnWaiting) - requested.CancelRequestedAt = time.Unix(1700000000, 0) - for name, test := range map[string]struct { - turn Turn - want turnCancellation - }{ - "queued": {turnWith(TurnQueued), cancellationEndsTurn}, - "waiting": {turnWith(TurnWaiting), cancellationReportsActivity}, - "waiting requested": {requested, cancellationRequested}, - "in progress": {turnWith(TurnInProgress), cancellationRequested}, - } { - if got := decideCancellation(test.turn); got != test.want { - t.Errorf("%s: %v", name, got) - } - } -} - -// CancelTurn decides from the Turn read before the request, and settles with -// the Turn read after it. +// CancelTurn ends a queued Turn with the Turn read after the request, and only +// records the request for a Turn that started. func TestCancelTurn(t *testing.T) { usage := json.RawMessage(`{"input_tokens":1}`) t.Run("queued ends cancelled", func(t *testing.T) { @@ -43,22 +24,9 @@ func TestCancelTurn(t *testing.T) { t.Fatalf("end %+v", ends) } }) - t.Run("waiting reports activity", func(t *testing.T) { - cancelling := turnWith(TurnWaiting) - cancelling.CancelRequestedAt = time.Unix(1700000001, 0) - var changes []SessionChange - f := &fakeTx{t: t, requestTurnCancel: done, loadTurn: returns(cancelling), loadUsage: returns(usage), appendChanges: collect(&changes)} - if err := CancelTurn(t.Context(), f, turnWith(TurnWaiting)); err != nil { - t.Fatal(err) - } - assertCalls(t, f, "RequestTurnCancel "+testTurn, "LoadTurn "+testTurn, "LoadUsage", "AppendChanges agent.session.in_progress") - if !reflect.DeepEqual(changes, []SessionChange{ActivityChange(cancelling, usage, nil)}) { - t.Fatalf("changes %+v", changes) - } - }) - t.Run("in progress only requests", func(t *testing.T) { + t.Run("waiting only requests", func(t *testing.T) { f := &fakeTx{t: t, requestTurnCancel: done} - if err := CancelTurn(t.Context(), f, turnWith(TurnInProgress)); err != nil { + if err := CancelTurn(t.Context(), f, turnWith(TurnWaiting)); err != nil { t.Fatal(err) } assertCalls(t, f, "RequestTurnCancel "+testTurn) diff --git a/services/core/internal/sessions/execution_functions.go b/services/core/internal/sessions/execution_functions.go index 1cbe24700..ffb1e2c74 100644 --- a/services/core/internal/sessions/execution_functions.go +++ b/services/core/internal/sessions/execution_functions.go @@ -156,6 +156,9 @@ func (o *ExecutionOperations) RecordFunctionCall(ctx context.Context, tenant, se func (o *ExecutionOperations) PendingFunctionCalls(ctx context.Context, tenant, session, turn string) ([]FunctionCall, error) { calls := make([]FunctionCall, 0) err := o.storage.WithFunctionTurn(ctx, tenant, session, turn, func(ctx context.Context, tx FunctionTx, current Turn) error { + if !acceptsFunctionResults(current) { + return nil + } pending, err := tx.LoadPendingFunctionCalls(ctx, current.ID) calls = append(calls, pending...) return err diff --git a/services/core/internal/sessions/functions.go b/services/core/internal/sessions/functions.go index 91376775c..5653369e7 100644 --- a/services/core/internal/sessions/functions.go +++ b/services/core/internal/sessions/functions.go @@ -159,17 +159,18 @@ func AdmitFunctionResult(ctx context.Context, tx FunctionResultTx, input Functio // or its application receipt. type PendingFunctionCallsTx interface { // LoadPendingFunctionCalls reads the Turn's calls without an application - // receipt, in creation order, while the Turn is in progress or waiting and - // no cancellation was requested, and none otherwise. + // receipt, in creation order, while the Turn is in progress or waiting, and + // none otherwise. LoadPendingFunctionCalls(ctx context.Context, turn string) ([]FunctionCall, error) } // LoadRequiredActions returns the function calls a Turn requires the caller to // answer as public required actions: its pending calls while it accepts -// results, and none otherwise. +// results, or while it waits after a cancellation request and has not ended, +// and none otherwise. func LoadRequiredActions(ctx context.Context, tx PendingFunctionCallsTx, turn Turn) ([]v1.FunctionCallAction, error) { actions := make([]v1.FunctionCallAction, 0) - if !acceptsFunctionResults(turn) { + if !acceptsFunctionResults(turn) && turn.Status != TurnWaiting { return actions, nil } calls, err := tx.LoadPendingFunctionCalls(ctx, turn.ID) diff --git a/services/core/internal/sessions/functions_test.go b/services/core/internal/sessions/functions_test.go index 38726c9fa..246a8bb40 100644 --- a/services/core/internal/sessions/functions_test.go +++ b/services/core/internal/sessions/functions_test.go @@ -283,9 +283,10 @@ func TestAdmitFunctionResult(t *testing.T) { func TestLoadRequiredActions(t *testing.T) { calls := []FunctionCall{testCall, {CallID: "second", ExecutorCallID: "native-second", Name: "search", Arguments: json.RawMessage(`{}`), Result: json.RawMessage(`{"success":true}`)}} - for _, status := range []string{TurnInProgress, TurnWaiting} { + // A waiting Turn keeps its required actions after a cancellation request, until it ends. + for _, turn := range []Turn{turnWith(TurnInProgress), turnWith(TurnWaiting), cancelling(TurnWaiting)} { f := &fakeFunctionTx{fakeTx: &fakeTx{t: t}, loadPendingFunctionCalls: returns(calls)} - actions, err := LoadRequiredActions(t.Context(), f, turnWith(status)) + actions, err := LoadRequiredActions(t.Context(), f, turn) if err != nil || len(actions) != 2 { t.Fatal(actions, err) } @@ -300,7 +301,7 @@ func TestLoadRequiredActions(t *testing.T) { } } } - for _, turn := range []Turn{turnWith(TurnQueued), cancelling(TurnWaiting), turnWith(TurnCompleted)} { + for _, turn := range []Turn{turnWith(TurnQueued), cancelling(TurnInProgress), turnWith(TurnCompleted)} { f := &fakeFunctionTx{fakeTx: &fakeTx{t: t}} actions, err := LoadRequiredActions(t.Context(), f, turn) if err != nil || actions == nil || len(actions) != 0 { diff --git a/services/core/tests/integration/function_state_test.go b/services/core/tests/integration/function_state_test.go index 7d011da21..ad7aa2383 100644 --- a/services/core/tests/integration/function_state_test.go +++ b/services/core/tests/integration/function_state_test.go @@ -102,9 +102,10 @@ func TestFunctionStateCancellationAndTerminalCleanup(t *testing.T) { t.Fatal(err) } } - assertFunctionState(t, s, tenant, session.ID, sessions.TurnWaiting, 0) + // The Turn keeps waiting with its required action, and no activity is journaled, until it ends. + assertFunctionState(t, s, tenant, session.ID, sessions.TurnWaiting, 1) changes, err := sessionAdapter(s).ListSessionEvents(t.Context(), tenant, session.ID, before) - if err != nil || len(changes) != 1 || changes[0].Event.Type != "agent.session.in_progress" || len(changes[0].RequiredActions) != 0 || changes[0].Turn.CancelRequestedAt.IsZero() { + if err != nil || len(changes) != 0 { t.Fatal(changes, err) } }