From 2cb18575f63319c03febfa3489ae8e3598a502b7 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 19 Oct 2026 10:53:01 +0000 Subject: [PATCH 1/7] Let Subagents call function and MCP tools --- .../internal/agent/codex/executor_turn.go | 8 +- .../agent/codex/function_receipt_test.go | 14 +- .../internal/agent/codex/function_receipts.go | 15 +- apps/daemon/internal/agent/codex/functions.go | 66 ++++++-- .../internal/agent/codex/functions_test.go | 52 +++++- apps/daemon/internal/agent/codex/session.go | 4 +- .../internal/agent/codex/session_cancel.go | 2 +- .../internal/agent/codex/session_messages.go | 4 +- .../agent/codex/subagent_observations.go | 2 + .../agent/codex/subagent_tools_native_test.go | 149 ++++++++++++++++++ contracts/agents-api/execution-tools.md | 4 +- contracts/agents-api/harness-capabilities.md | 4 +- contracts/agents-api/index.md | 1 + contracts/agents-api/subagents.md | 2 +- contracts/agents-api/zh/execution-tools.md | 6 +- .../agents-api/zh/harness-capabilities.md | 4 +- contracts/agents-api/zh/index.md | 3 +- contracts/agents-api/zh/subagents.md | 4 +- docs/runtime-protocol.md | 2 + docs/zh/runtime-protocol.md | 4 +- internal/agentdaemon/proto/functions.go | 7 +- .../harnessconfig/builtin/admission_test.go | 18 +-- internal/harnessconfig/codex/declaration.go | 3 - internal/harnessconfig/mcode/declaration.go | 3 - services/core/internal/execution/functions.go | 1 + .../internal/sessions/execution_functions.go | 18 ++- .../sessions/execution_functions_test.go | 15 ++ services/core/internal/sessions/inputs.go | 3 + .../integration/function_execution_test.go | 46 +++++- 29 files changed, 395 insertions(+), 69 deletions(-) create mode 100644 apps/daemon/internal/agent/codex/subagent_tools_native_test.go diff --git a/apps/daemon/internal/agent/codex/executor_turn.go b/apps/daemon/internal/agent/codex/executor_turn.go index 60746293f..747cae1ad 100644 --- a/apps/daemon/internal/agent/codex/executor_turn.go +++ b/apps/daemon/internal/agent/codex/executor_turn.go @@ -43,7 +43,7 @@ func (s *Session) settleExecutorTurn(startErr error) { s.operationsClosed = true s.operationMu.Unlock() s.stopSteering() - s.stopFunctionCalls() + s.stopFunctionCalls(false) if !s.nativeSettled.Load() || startErr != nil || !s.rpc.Alive() { s.cancelFn() s.settlementErr = errors.Join(s.settlementErr, s.rpc.Close()) @@ -117,7 +117,7 @@ func (s *Session) cancelExecutorTurn(ctx context.Context) error { s.cancelReady = make(chan struct{}) go func() { defer close(s.cancelReady) - s.stopFunctionCalls() + s.stopFunctionCalls(false) turnID, active := s.stopSteering() if !s.terminal.Load() && active { if turnID == "" { @@ -156,8 +156,8 @@ func (s *Session) onServerRequest(method string, handler ServerRequestHandler) { ThreadID string `json:"threadId"` TurnID string `json:"turnId"` } - if json.Unmarshal(raw, &scope) != nil || !s.isRootThread(scope.ThreadID) || s.terminal.Load() || - scope.TurnID == "" || !s.isRootTurn(scope.ThreadID, scope.TurnID) { + if json.Unmarshal(raw, &scope) != nil || scope.ThreadID == "" || scope.TurnID == "" || + (s.subagents == nil || s.isRootThread(scope.ThreadID)) && !s.isRootTurn(scope.ThreadID, scope.TurnID) { return nil, errors.New("codex: request does not belong to the active turn") } } diff --git a/apps/daemon/internal/agent/codex/function_receipt_test.go b/apps/daemon/internal/agent/codex/function_receipt_test.go index fd138a009..63ce1f357 100644 --- a/apps/daemon/internal/agent/codex/function_receipt_test.go +++ b/apps/daemon/internal/agent/codex/function_receipt_test.go @@ -137,7 +137,7 @@ func TestFunctionReceiptResultIsFrozenAndSubmittedOnce(t *testing.T) { if err := s.SubmitFunctionResult(t.Context(), result); err == nil { t.Fatal("settled result admitted again") } - s.stopFunctionCalls() + s.stopFunctionCalls(false) } func TestFunctionReceiptShutdownDoesNotConfirmApplication(t *testing.T) { @@ -151,7 +151,7 @@ func TestFunctionReceiptShutdownDoesNotConfirmApplication(t *testing.T) { case "run cancellation": s.cancelFn() case "terminal": - s.stopFunctionCalls() + s.stopFunctionCalls(false) } select { case err := <-done: @@ -161,7 +161,7 @@ func TestFunctionReceiptShutdownDoesNotConfirmApplication(t *testing.T) { case <-time.After(time.Second): t.Fatal("shutdown did not settle receipt") } - s.stopFunctionCalls() + s.stopFunctionCalls(false) }) } } @@ -199,7 +199,7 @@ func TestFunctionReceiptDeadlineEndsUncertainExecution(t *testing.T) { if s.cancelCtx.Err() == nil { t.Fatal("uncertain native execution remains active") } - s.stopFunctionCalls() + s.stopFunctionCalls(false) } // The receipt and deadline are both ready when the waiter resumes. Either select @@ -211,7 +211,7 @@ func TestFunctionReceiptWinsSimultaneousDeadline(t *testing.T) { s.setThreadID("thread") s.startSteering("thread", "turn") s.functions, _ = prepareFunctionTools(nil) - pending := &pendingFunction{turnID: "turn", name: "lookup", reply: &functionReply{Success: true, ContentItems: []functionContent{}}, receipt: make(chan error, 1)} + pending := &pendingFunction{thread: "thread", turnID: "turn", name: "lookup", reply: &functionReply{Success: true, ContentItems: []functionContent{}}, receipt: make(chan error, 1)} s.functions.pending["call"] = pending s.confirmFunctionResult(nativeFunctionReceipt("thread", "turn", "call", "lookup", "true", "completed", "[]")) ctx, cancel := context.WithCancel(t.Context()) @@ -254,11 +254,11 @@ func TestFunctionReceiptSurvivesObservationBackpressure(t *testing.T) { cancel() case "cancel": s.cancelled.Store(true) - s.stopFunctionCalls() + s.stopFunctionCalls(false) s.cancelFn() case "terminal": s.terminal.Store(true) - s.stopFunctionCalls() + s.stopFunctionCalls(false) } select { case err := <-receipt: diff --git a/apps/daemon/internal/agent/codex/function_receipts.go b/apps/daemon/internal/agent/codex/function_receipts.go index ba6727b16..f9264de9f 100644 --- a/apps/daemon/internal/agent/codex/function_receipts.go +++ b/apps/daemon/internal/agent/codex/function_receipts.go @@ -18,8 +18,11 @@ type functionReply struct { } type pendingFunction struct { - rpcID any - turnID string + rpcID any + thread string + turnID string + // held is the Subagent call that waits for its child Turn fact. + held *proto.Envelope name string reply *functionReply receipt chan error @@ -51,7 +54,7 @@ func (s *Session) SubmitFunctionResult(ctx context.Context, result proto.Functio } s.functions.mu.Lock() pending := s.functions.pending[result.CallID] - if pending == nil || pending.reply != nil || s.functions.closed || s.cancelled.Load() || s.terminal.Load() { + if pending == nil || pending.reply != nil || s.functions.closed || s.cancelled.Load() || (pending.thread == s.currentThreadID() && s.terminal.Load()) { s.functions.mu.Unlock() return agent.ErrUnknownFunctionCall } @@ -129,13 +132,13 @@ func (s *Session) confirmFunctionResult(raw json.RawMessage) { TurnID string `json:"turnId"` Item toolObservationSource `json:"item"` } - if json.Unmarshal(raw, &event) != nil || event.Item.Type != "dynamicToolCall" || !s.isRootTurn(event.ThreadID, event.TurnID) { + if json.Unmarshal(raw, &event) != nil || event.Item.Type != "dynamicToolCall" { return } s.functions.mu.Lock() defer s.functions.mu.Unlock() pending := s.functions.pending[event.Item.ID] - if pending == nil || pending.reply == nil || s.functions.closed || s.cancelled.Load() { + if pending == nil || pending.thread != event.ThreadID || pending.turnID != event.TurnID || pending.reply == nil || s.functions.closed || s.cancelled.Load() { return } item := event.Item @@ -144,7 +147,7 @@ func (s *Session) confirmFunctionResult(raw json.RawMessage) { status = "failed" } var err error - if pending.turnID != event.TurnID || pending.name != item.Tool || item.Namespace != "" || item.Status != status || item.Success == nil || *item.Success != pending.reply.Success || item.ContentItems == nil || !reflect.DeepEqual(*item.ContentItems, pending.reply.ContentItems) { + if pending.name != item.Tool || item.Namespace != "" || item.Status != status || item.Success == nil || *item.Success != pending.reply.Success || item.ContentItems == nil || !reflect.DeepEqual(*item.ContentItems, pending.reply.ContentItems) { err = errors.New("codex: native function receipt does not match submitted result") } delete(s.functions.pending, item.ID) diff --git a/apps/daemon/internal/agent/codex/functions.go b/apps/daemon/internal/agent/codex/functions.go index 31e6fdc5e..b58ba8145 100644 --- a/apps/daemon/internal/agent/codex/functions.go +++ b/apps/daemon/internal/agent/codex/functions.go @@ -1,6 +1,7 @@ package codex import ( + "context" "encoding/json" "errors" "strings" @@ -51,21 +52,38 @@ func (s *Session) handleFunctionCall(raw json.RawMessage, rpcID any) (any, error if err := json.Unmarshal(raw, &call); err != nil { return nil, err } - if s.functions == nil || !s.functions.names[call.Tool] || call.Namespace != nil || call.CallID == "" || !s.isRootTurn(call.ThreadID, call.TurnID) || !json.Valid(call.Arguments) { + root := s.isRootTurn(call.ThreadID, call.TurnID) + child := !root && s.subagents != nil && call.TurnID != "" && call.ThreadID != "" && !s.isRootThread(call.ThreadID) + if s.functions == nil || !s.functions.names[call.Tool] || call.Namespace != nil || call.CallID == "" || !(root || child) || !json.Valid(call.Arguments) { return nil, errors.New("unexpected function call") } + payload := proto.FunctionCallPayload{CallID: call.CallID, Name: call.Tool, Arguments: call.Arguments} + if child { + payload.NativeID, payload.TurnID = call.ThreadID, call.TurnID + } + env, err := proto.NewEnvelope(proto.TypeFunctionCall, s.runID, payload) + if err != nil { + return nil, err + } s.functions.mu.Lock() - if s.functions.closed || s.cancelled.Load() || s.terminal.Load() || s.functions.pending[call.CallID] != nil || len(s.functions.pending) >= 64 { + if s.functions.closed || s.cancelled.Load() || (root && s.terminal.Load()) || s.functions.pending[call.CallID] != nil || len(s.functions.pending) >= 64 { s.functions.mu.Unlock() return nil, errors.New("function call cannot be admitted") } - s.functions.pending[call.CallID] = &pendingFunction{rpcID: rpcID, turnID: call.TurnID, name: call.Tool, receipt: make(chan error, 1)} + pending := &pendingFunction{rpcID: rpcID, thread: call.ThreadID, turnID: call.TurnID, name: call.Tool, receipt: make(chan error, 1)} + if child { + pending.held = &env + } + s.functions.pending[call.CallID] = pending s.functions.mu.Unlock() - env, err := proto.NewEnvelope(proto.TypeFunctionCall, s.runID, proto.FunctionCallPayload{CallID: call.CallID, Name: call.Tool, Arguments: call.Arguments}) - if err == nil { - err = s.sendFunctionCall(env) + if child { + select { + case s.subagents.wake <- struct{}{}: + default: + } + return DeferReply, nil } - if err != nil { + if err = s.sendFunctionCall(env); err != nil { s.functions.mu.Lock() delete(s.functions.pending, call.CallID) s.functions.mu.Unlock() @@ -98,14 +116,38 @@ type functionContent struct { ImageURL *string `json:"imageUrl,omitempty"` } -func (s *Session) stopFunctionCalls() { +// stopFunctionCalls ends every pending call. With rootOnly it ends the root's +// calls and keeps accepting those of Subagents, which outlive the root Turn. +func (s *Session) stopFunctionCalls(rootOnly bool) { if s.functions != nil { s.functions.mu.Lock() - s.functions.closed = true - for _, pending := range s.functions.pending { - pending.receipt <- agent.ErrUnknownFunctionCall + s.functions.closed = !rootOnly + for id, pending := range s.functions.pending { + if !rootOnly || pending.thread == s.currentThreadID() { + pending.receipt <- agent.ErrUnknownFunctionCall + delete(s.functions.pending, id) + } } - clear(s.functions.pending) s.functions.mu.Unlock() } } + +// sendHeldFunctionCalls sends the Subagent calls whose child Turn Core already +// has, because Core resolves a child's call through its identity. +func (s *Session) sendHeldFunctionCalls(ctx context.Context) { + if s.functions == nil { + return + } + var ready []proto.Envelope + s.functions.mu.Lock() + for _, pending := range s.functions.pending { + if _, sent := s.subagents.sent[proto.TypeSubagentTurn+":"+pending.thread+":"+pending.turnID]; sent && pending.held != nil { + ready = append(ready, *pending.held) + pending.held = nil + } + } + s.functions.mu.Unlock() + for _, env := range ready { + s.sendWithin(ctx, env) + } +} diff --git a/apps/daemon/internal/agent/codex/functions_test.go b/apps/daemon/internal/agent/codex/functions_test.go index a2fd9c752..1c86ee6f2 100644 --- a/apps/daemon/internal/agent/codex/functions_test.go +++ b/apps/daemon/internal/agent/codex/functions_test.go @@ -109,7 +109,7 @@ func TestFunctionCallRejectsUnregisteredAndClosedRuns(t *testing.T) { t.Fatal("unexpected call accepted", params) } } - s.stopFunctionCalls() + s.stopFunctionCalls(false) if _, err := s.handleFunctionCall(json.RawMessage(`{"threadId":"thread","turnId":"turn","callId":"a","tool":"lookup","arguments":{}}`), "rpc"); err == nil { t.Fatal("closed run accepted call") } @@ -118,6 +118,56 @@ func TestFunctionCallRejectsUnregisteredAndClosedRuns(t *testing.T) { } } +func TestSubagentFunctionCallWaitsForChildTurnAndOutlivesRoot(t *testing.T) { + tc, srv, cleanup := NewTestClient() + defer cleanup() + s, out := newFunctionTestSession(tc.JSONRPCClient) + s.subagents = &subagentObservations{wake: make(chan struct{}, 1), sent: map[string]string{}} + s.functions, _ = prepareFunctionTools([]proto.FunctionTool{{Name: "lookup", Parameters: json.RawMessage(`{"type":"object"}`)}}) + s.setThreadID("root") + s.startSteering("root", "turn") + if err := SendServerRequest(srv, "rpc-call", "item/tool/call", map[string]any{"threadId": "child", "turnId": "child-turn", "callId": "call", "tool": "lookup", "arguments": map[string]string{}}); err != nil { + t.Fatal(err) + } + select { + case <-s.subagents.wake: + case <-time.After(time.Second): + t.Fatal("child call did not wake the observer") + } + // The root Turn ends while the child still waits for its answer. + s.terminal.Store(true) + s.stopFunctionCalls(true) + s.sendHeldFunctionCalls(t.Context()) + select { + case env := <-out: + t.Fatal("call sent before its child Turn fact", env) + default: + } + s.subagents.sent[proto.TypeSubagentTurn+":child:child-turn"] = "{}" + s.sendHeldFunctionCalls(t.Context()) + var call proto.FunctionCallPayload + if env := <-out; env.DecodePayload(&call) != nil || call.CallID != "call" || call.NativeID != "child" || call.TurnID != "child-turn" { + t.Fatal(env) + } + text := "answer" + finished := make(chan error, 1) + go func() { + finished <- s.SubmitFunctionResult(t.Context(), proto.FunctionResultPayload{CallID: "call", Success: true, Content: []proto.InputContent{{Type: "input_text", Text: &text}}}) + }() + var reply struct { + Result map[string]any `json:"result"` + } + if err := json.NewDecoder(srv.FromClient).Decode(&reply); err != nil { + t.Fatal(err) + } + reply.Result["type"], reply.Result["id"], reply.Result["tool"], reply.Result["status"] = "dynamicToolCall", "call", "lookup", "completed" + native, _ := json.Marshal(map[string]any{"threadId": "child", "turnId": "child-turn", "item": reply.Result}) + s.onItemCompleted(native) + if err := <-finished; err != nil { + t.Fatal(err) + } +} + func TestFunctionToolDefinitionsRejectAmbiguousInput(t *testing.T) { for _, tools := range [][]proto.FunctionTool{ {{Name: "", Parameters: json.RawMessage(`{}`)}}, diff --git a/apps/daemon/internal/agent/codex/session.go b/apps/daemon/internal/agent/codex/session.go index 71f6a1f7d..e93137552 100644 --- a/apps/daemon/internal/agent/codex/session.go +++ b/apps/daemon/internal/agent/codex/session.go @@ -336,7 +336,7 @@ func (s *Session) emitDoneAt(content string, usage *TurnUsage, completedAt *int6 return } s.stopSteering() - s.stopFunctionCalls() + s.stopFunctionCalls(s.subagents != nil) doneMeta := map[string]any{} if tid := s.currentThreadID(); tid != "" { doneMeta[proto.DoneMetaAgentSessionID] = tid @@ -376,7 +376,7 @@ func (s *Session) emitTerminalFailure(message string, asError bool, failure prot return } s.stopSteering() - s.stopFunctionCalls() + s.stopFunctionCalls(s.subagents != nil) // Always log: this is the only place the daemon decides "the prompt is // over, here's what went wrong (if anything)". Without this, post- // mortem requires correlating server-side TypeError frames against diff --git a/apps/daemon/internal/agent/codex/session_cancel.go b/apps/daemon/internal/agent/codex/session_cancel.go index 091eae119..a8e30ae65 100644 --- a/apps/daemon/internal/agent/codex/session_cancel.go +++ b/apps/daemon/internal/agent/codex/session_cancel.go @@ -40,7 +40,7 @@ func (s *Session) Cancel(ctx context.Context) error { func (s *Session) cancelNativeWork() { defer close(s.cancelReady) - s.stopFunctionCalls() + s.stopFunctionCalls(false) turnID, active := s.stopSteering() // Best effort: a known Turn must use its native identity. An explicit // empty ID invokes native startup cancellation before turn/started. diff --git a/apps/daemon/internal/agent/codex/session_messages.go b/apps/daemon/internal/agent/codex/session_messages.go index 4e1061dc6..7173f9626 100644 --- a/apps/daemon/internal/agent/codex/session_messages.go +++ b/apps/daemon/internal/agent/codex/session_messages.go @@ -50,11 +50,11 @@ func (s *Session) onItemCompleted(raw json.RawMessage) { if err := json.Unmarshal(raw, &p); err != nil { return } + // Record native application before observation backpressure can delay publication. + s.confirmFunctionResult(raw) if !s.isRootTurn(p.ThreadID, p.TurnID) || p.Item.ID == "" { return } - // Record native application before observation backpressure can delay publication. - s.confirmFunctionResult(raw) envs, text, err := DispatchCompletedItem(s.runID, p.Item, s.bufs) if err != nil { s.cfg.logger.Warn("codex: dispatch completed item failed", "run_id", s.runID, "err", err) diff --git a/apps/daemon/internal/agent/codex/subagent_observations.go b/apps/daemon/internal/agent/codex/subagent_observations.go index 29d1dcaf3..dda5bd36d 100644 --- a/apps/daemon/internal/agent/codex/subagent_observations.go +++ b/apps/daemon/internal/agent/codex/subagent_observations.go @@ -113,6 +113,7 @@ func (s *Session) collectSubagentFacts() { return } firstFailure = time.Time{} + s.sendHeldFunctionCalls(o.ctx) if terminal != nil && !busy { s.publishSubagentTerminal(terminal, nil) return @@ -129,6 +130,7 @@ func (s *Session) publishSubagentTerminal(events []proto.Envelope, cause error) for _, event := range events { s.trySend(event) } + s.stopFunctionCalls(false) s.closeOut() } diff --git a/apps/daemon/internal/agent/codex/subagent_tools_native_test.go b/apps/daemon/internal/agent/codex/subagent_tools_native_test.go new file mode 100644 index 000000000..bb5e37353 --- /dev/null +++ b/apps/daemon/internal/agent/codex/subagent_tools_native_test.go @@ -0,0 +1,149 @@ +package codex + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" +) + +// This opt-in test runs the real pinned harness with a model that spawns a +// Subagent, which calls a function or an MCP tool of the Agent. +func TestSubagentToolsNative(t *testing.T) { + if os.Getenv("OAC_TEST_CODEX_EXECUTOR_NATIVE") != "1" { + t.Skip("requires an authorized native model acceptance environment") + } + binary, root := os.Getenv("OAC_TEST_CODEX_BINARY"), os.Getenv("OAC_TEST_CODEX_LIVE_ROOT") + model, endpoint := os.Getenv("OAC_TEST_CODEX_MODEL"), os.Getenv("OAC_TEST_CODEX_BASE_URL") + key, err := os.ReadFile(os.Getenv("OAC_TEST_CODEX_KEY_FILE")) + if binary == "" || root == "" || model == "" || endpoint == "" || err != nil { + t.Fatal("missing explicit native acceptance configuration") + } + isolated, err := os.MkdirTemp(root, "run-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(isolated) }) + t.Setenv("OAC_RUNTIME_HOME", isolated) + mcpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var rpc struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + _ = json.Unmarshal(body, &rpc) + var result string + switch rpc.Method { + case "initialize": + result = `{"protocolVersion":"2025-06-18","capabilities":{"tools":{}},"serverInfo":{"name":"proof","version":"0"}}` + case "tools/list": + result = `{"tools":[{"name":"echo","description":"Return the synthetic MCP value.","inputSchema":{"type":"object","properties":{}}}]}` + case "tools/call": + result = `{"content":[{"type":"text","text":"MCP-5521"}]}` + default: + w.WriteHeader(http.StatusAccepted) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":` + string(rpc.ID) + `,"result":` + result + `}`)) + })) + defer mcpServer.Close() + for name, prompt := range map[string]string{ + "function": "Call spawn_agent with fork_context true and the task: call the lookup function once and report the value it returns. Then wait for that Subagent and reply with its value. Do not call lookup yourself.", + "mcp": "Call spawn_agent with a task: call the echo tool of the proof MCP server once and report the value it returns. Then wait for that Subagent and reply with its value. Do not call echo yourself.", + } { + t.Run(name, func(t *testing.T) { + cfg := defaultSessionConfig(testStateRoot(t)) + cfg.codexBinary = binary + cfg.logger = slog.New(slog.DiscardHandler) + servers := []proto.MCPHTTPServer{{ConnectionOrigin: "service", ServerLabel: "proof", ServerURL: mcpServer.URL}} + req := proto.PromptRequestPayload{ + AgentKind: "codex", AgentStateKey: "subagent-tools-" + name, + DisableExecutionEnvironment: true, ObserveSubagentIdentities: true, + AgentOptions: map[string]any{"model": model, "model_provider": map[string]any{"base_url": endpoint, "protocol": "responses", "api_key": strings.TrimSpace(string(key))}}, + FunctionTools: []proto.FunctionTool{{Name: "lookup", Description: "Return the synthetic verification value.", Parameters: json.RawMessage(`{"type":"object","properties":{},"additionalProperties":false}`)}}, + MCPHTTPServers: &servers, + } + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Minute) + defer cancel() + e, err := newExecutor(ctx, req, cfg) + if err != nil { + t.Fatalf("native prepare failed (%T)", err) + } + defer e.Close(context.Background()) + output := make(chan proto.Envelope, 512) + turn, err := e.StartTurn(ctx, name, proto.TextInput(prompt), output) + if err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var calls []proto.FunctionCallPayload + var childTools []string + var answer string + var answers sync.WaitGroup + for event := range output { + switch event.Type { + case proto.TypeFunctionCall: + var call proto.FunctionCallPayload + _ = event.DecodePayload(&call) + mu.Lock() + calls = append(calls, call) + mu.Unlock() + text := "FN-4729" + answers.Add(1) + go func() { + defer answers.Done() + if err := turn.(agent.FunctionResultSubmitter).SubmitFunctionResult(ctx, proto.FunctionResultPayload{CallID: call.CallID, Success: true, Content: []proto.InputContent{{Type: "input_text", Text: &text}}}); err != nil { + t.Errorf("function answer: %v", err) + } + }() + case proto.TypeSubagentItem: + var item proto.SubagentItemPayload + _ = event.DecodePayload(&item) + var tool struct { + Stage string `json:"stage"` + Observation struct { + Kind, Name, Server string + } `json:"observation"` + } + if item.Kind == proto.TypeToolCall && json.Unmarshal(item.Payload, &tool) == nil && tool.Stage == "after" { + childTools = append(childTools, tool.Observation.Kind+":"+tool.Observation.Server+tool.Observation.Name) + } + case proto.TypeDone: + var done proto.DonePayload + _ = event.DecodePayload(&done) + answer = done.Content + } + } + answers.Wait() + if _, err := turn.AwaitSettlement(ctx); err != nil { + t.Fatal(err) + } + t.Logf("function_calls=%+v child_tools=%v answer=%q", calls, childTools, answer) + want := map[string]string{"function": "function:lookup", "mcp": "mcp:proofecho"}[name] + seen := false + for _, tool := range childTools { + seen = seen || tool == want + } + if !seen { + t.Fatalf("the Subagent did not run %s", want) + } + if name == "function" && (len(calls) != 1 || calls[0].NativeID == "" || calls[0].TurnID == "" || !strings.Contains(answer, "FN-4729")) { + t.Fatal("the Subagent call did not reach the client with its child identity") + } + if name == "mcp" && (len(calls) != 0 || !strings.Contains(answer, "MCP-5521")) { + t.Fatal("the MCP value did not reach the root") + } + }) + } +} diff --git a/contracts/agents-api/execution-tools.md b/contracts/agents-api/execution-tools.md index 756be7c71..e69ec24e0 100644 --- a/contracts/agents-api/execution-tools.md +++ b/contracts/agents-api/execution-tools.md @@ -25,7 +25,9 @@ A function declaration requires `name`, `description` and a JSON Schema in `para | Unknown call, or a call of another Turn, in the caller's Session | 400 `invalid_request_error`; the pending action is unchanged | | Missing or foreign Session | 404 | -[Session input conflicts](./sessions-events.md#input-errors) records the exact messages. Invalid or unsupported content cannot consume a pending call. Admission is separate from application: the adapter confirms a result only when the matching native tool result appears in the live root Turn ([receipt contract](./message-content.md#function-results)). A transport write alone confirms nothing, and a confirmation says nothing about provider consumption or exactly-once external effects. Core never replays a result automatically. +[Session input conflicts](./sessions-events.md#input-errors) records the exact messages. Invalid or unsupported content cannot consume a pending call. Admission is separate from application: the adapter confirms a result only when the matching native tool result appears in the live Turn that made the call ([receipt contract](./message-content.md#function-results)). A transport write alone confirms nothing, and a confirmation says nothing about provider consumption or exactly-once external effects. Core never replays a result automatically. + +A function call made by a Subagent is a required action of the root Turn that runs the child. Its `turn_id` is that root Turn, and its `call_id` equals the `call_id` of the child's `function_call` Item on the [Subagent routes](./subagents.md#public-reads). Answer it with `function_call_output` / `tool_result` under that `call_id` and the root Turn ID; the child's Turn ID is not a Session Turn and gets the 400 for a call of another Turn. A Subagent's HTTP MCP calls are `mcp_call` Items in the child's history. ### Required actions and recovery diff --git a/contracts/agents-api/harness-capabilities.md b/contracts/agents-api/harness-capabilities.md index 004ba060b..0b178d03a 100644 --- a/contracts/agents-api/harness-capabilities.md +++ b/contracts/agents-api/harness-capabilities.md @@ -45,9 +45,9 @@ Native model parameters and provider protocols per Harness are in [model executi | [HTTP MCP bearer credentials](./execution-tools.md#http-mcp) | Verified | Verified | Verified | | Required MCP initialization | Verified | Verified | Rejected | | [Subagents](./subagents.md) | Verified: hosted; admitted: `none`, self-hosted | Verified: hosted; admitted: `none`, self-hosted | Verified: hosted; admitted: `none`, self-hosted | -| Subagents together with functions or HTTP MCP | Rejected | Rejected | Rejected | +| Subagents together with functions or HTTP MCP | Verified: `none`; admitted: hosted, self-hosted | Rejected | HTTP MCP admitted; functions rejected for every Session | -Claude structured output requires medium verbosity and combines with `multi_agent`, Skills, Plugins, capability directories, MCP and `tool_search`. Claude deferred discovery combines with MCP, Skills, Plugins, capability directories and structured output. The Subagent profile does not carry function tools, so deferred discovery with `multi_agent` stays rejected. Claude Code decides whether `tool_search` takes effect for the configured model, provider and beta settings. Claude on a workspace placement needs the packaged bridge features for each operation (`workspace_functions`, `workspace_structured_output`, `workspace_tool_search`, `workspace_mcp_http`). +Claude structured output requires medium verbosity and combines with `multi_agent`, Skills, Plugins, capability directories, MCP and `tool_search`. Claude deferred discovery combines with MCP, Skills, Plugins, capability directories and structured output. Claude Code does not carry function tools into the Subagent profile, so deferred discovery with `multi_agent` stays rejected. Claude Code decides whether `tool_search` takes effect for the configured model, provider and beta settings. Claude on a workspace placement needs the packaged bridge features for each operation (`workspace_functions`, `workspace_structured_output`, `workspace_tool_search`, `workspace_mcp_http`). ## Environment preparation diff --git a/contracts/agents-api/index.md b/contracts/agents-api/index.md index 49d094284..d82944228 100644 --- a/contracts/agents-api/index.md +++ b/contracts/agents-api/index.md @@ -151,6 +151,7 @@ Each item is Core's deliberate or native behavior where the official service beh ### Unverified against the official service - The order of errors when one request has several faults, and error, default and payload-limit parity in general. +- A Subagent's function call: Core makes it a required action of the root Turn that runs the child, with `turn_id` set to that Turn. The Turn is `waiting` and the Session `requires_action` while root output may continue, and the `call_id` equals the `call_id` of the child's `function_call` Item. A Subagent's MCP calls are `mcp_call` Items in the child's history. - Whether Subagent child Turns and pending Environment file writes block Session deletion. - The official order between the pending-input error and an unknown result target. - Failure reasons for npm, initial-file and Skill installation have no official sample. diff --git a/contracts/agents-api/subagents.md b/contracts/agents-api/subagents.md index 64d840fc8..be7eedc3f 100644 --- a/contracts/agents-api/subagents.md +++ b/contracts/agents-api/subagents.md @@ -50,7 +50,7 @@ An adapter freezes root output before child settlement, keeps its native owner a ## Native profiles -[Harness capabilities](./harness-capabilities.md#tools) lists rejected tool combinations. +A Subagent uses the Agent's function and HTTP MCP tools as the Harness would in a local run, and [Harness capabilities](./harness-capabilities.md#tools) lists each combination. A child's function call is a required action of the root Turn ([Functions](./execution-tools.md#functions)); a child's MCP call is an `mcp_call` Item in its history. Codex children inherit the Agent's HTTP MCP servers, and a forked child inherits its functions while a child with fresh history has none. MiniMax Code projects its children's MCP calls and rejects public functions for the Harness as a whole. Claude Code rejects functions and HTTP MCP with Subagents. **Codex.** The adapter enables the native `multi_agent` feature and maps the concurrency limit to `agents.max_threads`; nesting depth and other native features keep their Codex defaults. It refuses to start if managed requirements force code mode or `multi_agent_v2`, whose collaboration calls leave no persisted receipt. Child Items without a neutral Item type, such as image views and context compaction, are not projected. Close and reopen facts come from direct tool output correlated with the same call's persisted completion, so they need native persisted receipts. Native Turn times have second precision. Child file work can finish under the same owner after the root Turn finishes. Cancellation continues under the same owner after a caller deadline; a later call can confirm settlement without repeating the native interrupt. diff --git a/contracts/agents-api/zh/execution-tools.md b/contracts/agents-api/zh/execution-tools.md index 8faf7e25c..b346fd88f 100644 --- a/contracts/agents-api/zh/execution-tools.md +++ b/contracts/agents-api/zh/execution-tools.md @@ -1,7 +1,7 @@ --- title: "执行工具" source: contracts/agents-api/execution-tools.md -source_hash: ada5752faa24616f4fa75c6559028e3b24be0a4db92b77c7baf936d68021b8df +source_hash: 09fe0e89ed0d75dac206e08aff4297d3e8f1830a771a6faabc7277070f836546 --- Agent 在 `tools` 中声明应用函数、控制项和 MCP 服务器，并可在 `text.format` 中声明输出 schema。本契约说明 Core 如何验证声明、哪些内容跨越 Runtime 边界，以及调用方如何恢复待执行操作。[Harness 能力](harness-capabilities.md)列出各 Harness 在不同部署位置支持的操作。原生工作区工具和 Environment Plugin MCP 属于 [Environment](environments.md#skills-plugins-and-environment-mcp)。 @@ -27,7 +27,9 @@ Agent 在 `tools` 中声明应用函数、控制项和 MCP 服务器，并可在 | 调用方 Session 内未知调用或其他 Turn 的调用 | 400 `invalid_request_error`；待执行操作不变 | | Session 不存在或属于其他范围 | 404 | -[Session 输入冲突](sessions-events.md#input-errors)记录准确消息。无效或不支持的内容不能消耗待处理调用。准入与应用结果分离：只有匹配的原生工具结果出现在实时根 Turn 中，适配器才确认结果（[回执契约](message-content.md#function-results)）。传输写入本身不确认任何事实，确认也不代表提供商已消费结果或外部副作用恰好发生一次。Core 不自动重放结果。 +[Session 输入冲突](sessions-events.md#input-errors)记录准确消息。无效或不支持的内容不能消耗待处理调用。准入与应用结果分离：只有匹配的原生工具结果出现在发起调用的实时 Turn 中，适配器才确认结果（[回执契约](message-content.md#function-results)）。传输写入本身不确认任何事实，确认也不代表提供商已消费结果或外部副作用恰好发生一次。Core 不自动重放结果。 + +Subagent 发起的函数调用是运行该子级的根 Turn 的待执行操作：`turn_id` 是该根 Turn，`call_id` 等于 [Subagent 路由](subagents.md#public-reads)上该子级 `function_call` Item 的 `call_id`。使用该 `call_id` 和根 Turn ID，以 `function_call_output` / `tool_result` 作答；子级的 Turn ID 不是 Session Turn，会得到“另一个 Turn 的调用”的 400。Subagent 的 HTTP MCP 调用是该子级历史中的 `mcp_call` Item。 ### 待执行操作与恢复 {#required-actions-and-recovery} diff --git a/contracts/agents-api/zh/harness-capabilities.md b/contracts/agents-api/zh/harness-capabilities.md index 8abac57f9..26aade8be 100644 --- a/contracts/agents-api/zh/harness-capabilities.md +++ b/contracts/agents-api/zh/harness-capabilities.md @@ -1,7 +1,7 @@ --- title: "Harness 能力" source: contracts/agents-api/harness-capabilities.md -source_hash: 4a37a1f01a48dc0b9527de1edecb9117a2506558fcbefed2a297561edb5a4a07 +source_hash: 6ae7ae05948fe30ce66d86b0a3987c08456392d6a7ed66e71c09ec8b85044843 --- 本页列出每个 Harness 在每种部署位置支持的能力。Core 根据 `internal/harnessconfig//declaration.go` 中 Harness 的声明决定准入，运行 Session 的 Runtime 也必须声明操作。所链接契约定义各操作；[Harness 接入](harness-onboarding.md#qualify-the-adapter)说明资格验证方法。 @@ -47,7 +47,7 @@ source_hash: 4a37a1f01a48dc0b9527de1edecb9117a2506558fcbefed2a297561edb5a4a07 | [HTTP MCP bearer 凭据](execution-tools.md#http-mcp) | 已验证 | 已验证 | 已验证 | | 必需 MCP 初始化 | 已验证 | 已验证 | 已拒绝 | | [Subagent](subagents.md) | 已验证：托管；已准入：`none`、自托管 | 已验证：托管；已准入：`none`、自托管 | 已验证：托管；已准入：`none`、自托管 | -| Subagent 与函数或 HTTP MCP 同时使用 | 已拒绝 | 已拒绝 | 已拒绝 | +| Subagent 与函数或 HTTP MCP 同时使用 | 已验证：`none`；已准入：hosted、self-hosted | 已拒绝 | HTTP MCP 已准入；每个 Session 都拒绝函数 | Claude 结构化输出要求 medium verbosity，可与 `multi_agent`、Skill、Plugin、能力目录、MCP 和 `tool_search` 组合。Claude 延迟发现可与 MCP、Skill、Plugin、能力目录和结构化输出组合。Subagent 配置不携带函数工具，因此延迟发现与 `multi_agent` 的组合仍被拒绝。`tool_search` 是否对所配置的模型、提供商和 beta 设置生效，由 Claude Code 决定。Claude 在工作区位置需要每种操作的打包桥接功能（`workspace_functions`、`workspace_structured_output`、`workspace_tool_search`、`workspace_mcp_http`）。 diff --git a/contracts/agents-api/zh/index.md b/contracts/agents-api/zh/index.md index 75a0dc6bd..dc22ffe5b 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: 57fa589c761bdb16a56dce08b287a61a94d72fdc3ef7f33aacc10a1eb8ba0e62 +source_hash: f9e0c35869525870fa49e725a05bfda1d9b4d94af95764728575971ac7cce92e --- Core 实现下方固定版本的 OpenAI Agents API。本台账列出各项资源的覆盖情况、相对于官方服务的所有差异以及已知缺口。 @@ -153,6 +153,7 @@ Core 自身字段位于 `x_agents_core` 中（[Core extensions](../../../docs/zh ### 尚未与官方服务核实的内容 {#unverified-against-the-official-service} +- Subagent 的函数调用：Core 将其作为运行该子级的根 Turn 的待执行操作，`turn_id` 为该 Turn。该 Turn 为 `waiting`、Session 为 `requires_action`，根输出可继续，`call_id` 等于该子级 `function_call` Item 的 `call_id`。Subagent 的 MCP 调用是该子级历史中的 `mcp_call` Item。 - 当一个请求存在多项故障时的错误顺序，以及错误处理、默认值和载荷限制的总体一致性。 - Subagent 子级 Turns 和待处理的 Environment file 写入是否会阻止 Session 删除。 - 官方服务在待处理输入错误与未知结果目标之间的先后顺序。 diff --git a/contracts/agents-api/zh/subagents.md b/contracts/agents-api/zh/subagents.md index b78c93140..03fee3bfe 100644 --- a/contracts/agents-api/zh/subagents.md +++ b/contracts/agents-api/zh/subagents.md @@ -1,7 +1,7 @@ --- title: "子 Agent" source: contracts/agents-api/subagents.md -source_hash: 9257e25d1b4d10de9643157510f6695c0a16cc7ee1b21393cf9589f985c577f8 +source_hash: c70156da39cb9eac2a6e86e7b6807ef4fef0315c59c84cc84f506f8dcf905ec6 --- 启用 `multi_agent.enabled` 后，Harness 可以启动原生子 Agent。Core 通过锁定版本的 SDK（见 [`upstream.json`](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/contracts/agents-api/upstream.json)）中的六项 Subagent 读取操作公开它们，并根据适配器观察结果记录它们。当 `multi_agent.enabled=false` 时，Runtime 会移除原生子 Agent 工具。[Harness capabilities](harness-capabilities.md) 列出各 Harness 在哪些组合中支持子 Agent。 @@ -52,7 +52,7 @@ Core 根据已授权的 Session 绑定关系分配合公开 ID 和所有权。 ## 原生配置 {#native-profiles} -[Harness capabilities](harness-capabilities.md#tools) 列出被拒绝的工具组合。 +Subagent 像本地运行该 Harness 一样使用 Agent 的函数和 HTTP MCP 工具，各组合见 [Harness capabilities](harness-capabilities.md#tools)。子级的函数调用是根 Turn 的待执行操作（[Functions](execution-tools.md#functions)）；子级的 MCP 调用是其历史中的 `mcp_call` Item。Codex 子级继承 Agent 的 HTTP MCP 服务器；fork 出的子级继承其函数，使用全新历史的子级没有函数。MiniMax Code 会投影其子级的 MCP 调用，并对整个 Harness 拒绝公开函数。Claude Code 在使用 Subagent 时拒绝函数和 HTTP MCP。 **Codex.** 适配器会启用原生 `multi_agent` 特性，并把并发限制映射到 `agents.max_threads`；嵌套深度和其他原生特性保持 Codex 默认值。如果托管配置要求强制启用代码模式或 `multi_agent_v2`（它们的协作调用不留下持久化回执），则拒绝启动。没有中立 Item 类型的子级 Item（例如图像查看和上下文压缩）不会被投影。关闭和重新打开的事实来自直接工具输出，并与同一次调用的持久化完成记录相关联，因此需要原生持久化回执。原生 Turn 时间具有秒级精度。根级 Turn 完成后，子级文件工作可以在同一所有者下完成。超过调用方截止时间后，取消操作仍会在同一所有者下继续；后续调用可以确认已经收敛，而无需重复执行原生中断。 diff --git a/docs/runtime-protocol.md b/docs/runtime-protocol.md index b65ca49d3..e739e3d56 100644 --- a/docs/runtime-protocol.md +++ b/docs/runtime-protocol.md @@ -116,6 +116,8 @@ The linked source files define the required fields, validators, limits and finit Initial, prepared and active input use the same [ordered MessageInput](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/message_input.go). Adapters keep message and content order and reject unsupported content explicitly; a text-only transport rejects image content rather than dropping it. The [message input contract](../contracts/agents-api/message-content.md) owns the public image profile, whitespace rules and each Harness's native conversion. +A `function_call` carries `native_id` and `turn_id` when a Subagent made the call, with the meaning they have in `subagent_turn`; both are empty for a call of the Run itself. The adapter sends the child's `subagent_identity` and `subagent_turn` before the call, and Core answers with `function_result` under the same `call_id`. + Usage frames and the final usage snapshot each carry the cumulative measurement of the current execution and replace the previous snapshot; never add them. An absent measurement is unknown, not zero. ## Preparation and execution order diff --git a/docs/zh/runtime-protocol.md b/docs/zh/runtime-protocol.md index c73ac4475..bc60a3897 100644 --- a/docs/zh/runtime-protocol.md +++ b/docs/zh/runtime-protocol.md @@ -1,7 +1,7 @@ --- title: "Core–Runtime 协议" source: docs/runtime-protocol.md -source_hash: 5f735aa86ed4ac1e82cfe028eb10eba50d651df844c10d73f739e5764fc6dc82 +source_hash: dfd5005d5d5a51df20d1407fe897237302c7e16f0ad6e6decd0910d181e3f039 --- Core–Runtime 协议连接 Core 与 Runtime daemon，定义 daemon 连接上消息的含义和顺序。托管和自托管 Runtime 使用同一协议。 @@ -118,6 +118,8 @@ Core 对 `usage` 和 `resume` 没有准入规则。 初始、已准备和活动输入使用同一[有序 MessageInput](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/message_input.go)。Adapter 保留消息与内容顺序，并明确拒绝不支持的内容；仅文本 transport 拒绝图像内容而不丢弃它。[消息输入契约](../../contracts/agents-api/zh/message-content.md)负责公开图像 profile、空白规则和各 Harness 的原生转换。 +Subagent 发起调用时，`function_call` 携带 `native_id` 和 `turn_id`，含义与 `subagent_turn` 中相同；Run 自身发起的调用两者均为空。Adapter 在调用之前发送该子级的 `subagent_identity` 和 `subagent_turn`，Core 用同一个 `call_id` 的 `function_result` 作答。 + Usage frame 和最终 usage snapshot 都携带当前执行的累计测量，替换之前的快照；不要相加。缺失的测量表示未知，不是零。 ## 准备与执行顺序 {#preparation-and-execution-order} diff --git a/internal/agentdaemon/proto/functions.go b/internal/agentdaemon/proto/functions.go index b678c135c..d9acd9764 100644 --- a/internal/agentdaemon/proto/functions.go +++ b/internal/agentdaemon/proto/functions.go @@ -36,9 +36,14 @@ func (r PromptRequestPayload) ValidateToolSearch(supported bool) error { return nil } -// FunctionCallPayload belongs to the Run identified by Envelope.ID. +// FunctionCallPayload belongs to the Run identified by Envelope.ID. NativeID +// and TurnID name the native child and its Turn when a Subagent made the call, +// with the meaning they have in SubagentTurnPayload; both are empty for a call +// of the Run itself. type FunctionCallPayload struct { CallID string `json:"call_id"` + NativeID string `json:"native_id,omitempty"` + TurnID string `json:"turn_id,omitempty"` Name string `json:"name"` Arguments json.RawMessage `json:"arguments"` } diff --git a/internal/harnessconfig/builtin/admission_test.go b/internal/harnessconfig/builtin/admission_test.go index 65a8b1629..00b46c53d 100644 --- a/internal/harnessconfig/builtin/admission_test.go +++ b/internal/harnessconfig/builtin/admission_test.go @@ -42,6 +42,9 @@ func TestSubagentDeclarationsPreserveSupportedBoundaries(t *testing.T) { for _, kind := range []string{"claude_sdk", "codex", "mcode"} { t.Run(kind, func(t *testing.T) { declaration, _ := Registry().Lookup(kind) + if declaration.ValidateConfiguration == nil { + return // Only the common checks apply. + } admission := harnessconfig.Admission{MultiAgent: true} for _, placement := range []string{"none", "openai_hosted", "self_hosted"} { admission.Placement = placement @@ -49,21 +52,12 @@ func TestSubagentDeclarationsPreserveSupportedBoundaries(t *testing.T) { t.Fatal(placement, err) } } - // Public functions and MCP stay independent capabilities: Subagent - // support does not admit their combinations. - if kind == "mcode" { - return - } // Its tool declaration rejects both for every Session. + // Codex and MiniMax Code admit function and MCP tools with Subagents. admission.Placement = "none" for _, tool := range []string{`{"type":"function","name":"f"}`, `{"type":"mcp","server_label":"s"}`} { admission.Tools = []json.RawMessage{json.RawMessage(tool)} - admission.MultiAgent = true - if err := declaration.ValidateConfiguration(admission); err == nil { - t.Fatal("unsupported multi-agent combination accepted", tool) - } - admission.MultiAgent = false - if err := declaration.ValidateConfiguration(admission); err != nil { - t.Fatal("single-agent declaration changed", err) + if err := declaration.ValidateConfiguration(admission); (err != nil) != (kind == "claude_sdk") { + t.Fatal(kind, tool, err) } } }) diff --git a/internal/harnessconfig/codex/declaration.go b/internal/harnessconfig/codex/declaration.go index 0e4e2af4b..5b708aaa6 100644 --- a/internal/harnessconfig/codex/declaration.go +++ b/internal/harnessconfig/codex/declaration.go @@ -69,9 +69,6 @@ var Declaration = harnessconfig.Declaration{ return rejectRemoteImages(proto.MessageInput{{Content: result.Content}}) }, ValidateConfiguration: func(admission harnessconfig.Admission) error { - if err := harnessconfig.RejectSubagentTools(admission, "function", "mcp"); err != nil { - return err - } // Subagent observation turns native code_mode off. if admission.MultiAgent && enablesProgrammaticToolCalling(admission.Tools) { return errors.New("programmatic_tool_calling.enabled=true is not supported with multi_agent by this Harness.") diff --git a/internal/harnessconfig/mcode/declaration.go b/internal/harnessconfig/mcode/declaration.go index 6667060ae..35275d33b 100644 --- a/internal/harnessconfig/mcode/declaration.go +++ b/internal/harnessconfig/mcode/declaration.go @@ -63,9 +63,6 @@ var Declaration = harnessconfig.Declaration{ Placements: []string{"none", "openai_hosted", "self_hosted"}, MCPOrigins: []string{"service", "environment"}, WhitespaceOnlyText: proto.CapabilityUnsupported, - ValidateConfiguration: func(admission harnessconfig.Admission) error { - return harnessconfig.RejectSubagentTools(admission, "mcp") - }, ValidateTools: func(functions []proto.FunctionTool, mcp []proto.MCPHTTPServer) error { if len(functions) > 0 { return errors.New("The configured engine does not support public functions.") diff --git a/services/core/internal/execution/functions.go b/services/core/internal/execution/functions.go index 2fcb5b2e4..39c1d9bf3 100644 --- a/services/core/internal/execution/functions.go +++ b/services/core/internal/execution/functions.go @@ -69,6 +69,7 @@ func (f *functionExchange) record(ctx context.Context, env proto.Envelope) error } err := f.sessions.RecordFunctionCall(ctx, f.tenant, f.session, f.turn, sessions.FunctionCall{ CallID: items.Identity(f.turn, "tool:"+call.CallID), ExecutorCallID: call.CallID, Name: call.Name, Arguments: call.Arguments, + Subagent: call.NativeID, SubagentTurn: call.TurnID, }) return f.unlessCancelling(ctx, err) } diff --git a/services/core/internal/sessions/execution_functions.go b/services/core/internal/sessions/execution_functions.go index 1285a1a9e..a39d49e6d 100644 --- a/services/core/internal/sessions/execution_functions.go +++ b/services/core/internal/sessions/execution_functions.go @@ -3,6 +3,8 @@ package sessions import ( "context" "encoding/json" + + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/items" ) // FunctionCallMatch is what a Turn's stored call shows about a reported call. @@ -20,6 +22,9 @@ type FunctionCallMatch struct { type FunctionTx interface { JournalTx PendingFunctionCallsTx + // LoadNativeSubagent reads the Session's native Subagent and reports + // whether the Session has it. + LoadNativeSubagent(ctx context.Context, native string) (NativeSubagent, bool, error) // MatchFunctionCall compares call with the Turn's call of the same public // identity. MatchFunctionCall(ctx context.Context, turn string, call FunctionCall) (FunctionCallMatch, error) @@ -76,13 +81,24 @@ func decideFunctionConfirmation(turn Turn, call FunctionCall) (bool, error) { } // RecordFunctionCall records a call an executor reported for its Turn and -// journals the Turn's required actions with it. A retry of the recorded call +// journals the Turn's required actions with it. A call a Subagent made takes +// the identity of the child's function Item. A retry of the recorded call // changes nothing. func (o *ExecutionOperations) RecordFunctionCall(ctx context.Context, tenant, session, turn string, call FunctionCall) error { if err := validateFunctionCall(call); err != nil { return err } return o.storage.WithFunctionTurn(ctx, tenant, session, turn, func(ctx context.Context, tx FunctionTx, current Turn) error { + if call.Subagent != "" || call.SubagentTurn != "" { + child, found, err := tx.LoadNativeSubagent(ctx, call.Subagent) + if err != nil { + return err + } + if !found || !child.Visible || !validNativeIdentity(call.SubagentTurn) { + return ErrInvalidInput + } + call.CallID = items.Identity(items.Identity(child.ID, "turn:"+call.SubagentTurn), "tool:"+call.ExecutorCallID) + } match, err := tx.MatchFunctionCall(ctx, current.ID, call) if err != nil { return err diff --git a/services/core/internal/sessions/execution_functions_test.go b/services/core/internal/sessions/execution_functions_test.go index 03cdbb09e..62deff254 100644 --- a/services/core/internal/sessions/execution_functions_test.go +++ b/services/core/internal/sessions/execution_functions_test.go @@ -8,6 +8,7 @@ import ( "testing" v1 "github.com/MiniMax-AI/OpenAgentCore/contracts/agents-api/v1" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/items" ) // fakeExecutionStorage is strict lease-bound Session storage: a method whose @@ -180,6 +181,20 @@ func TestRecordFunctionCall(t *testing.T) { assertCalls(t, f.fakeTx, "MatchFunctionCall "+testTurn+" call") } }) + t.Run("a Subagent call takes the identity of the child's function Item", func(t *testing.T) { + call := testCall + call.Subagent, call.SubagentTurn = "child", "child-turn" + hidden := &fakeFunctionTx{fakeTx: &fakeTx{t: t, loadNativeSubagent: native(NativeSubagent{ID: testSubagent})}} + if err := functionOperations(t, hidden, turnWith(TurnInProgress)).RecordFunctionCall(t.Context(), testTenant, testSession, testTurn, call); !errors.Is(err, ErrInvalidInput) { + t.Errorf("hidden child: %v", err) + } + want := items.Identity(items.Identity(testSubagent, "turn:child-turn"), "tool:native-call") + f := &fakeFunctionTx{fakeTx: &fakeTx{t: t, loadNativeSubagent: native(NativeSubagent{ID: testSubagent, Visible: true})}, matchFunctionCall: returns(FunctionCallMatch{Recorded: true, Matches: true})} + if err := functionOperations(t, f, turnWith(TurnWaiting)).RecordFunctionCall(t.Context(), testTenant, testSession, testTurn, call); err != nil { + t.Fatal(err) + } + assertCalls(t, f.fakeTx, "LoadNativeSubagent child", "MatchFunctionCall "+testTurn+" "+want) + }) t.Run("an invalid call reaches no storage", func(t *testing.T) { call := testCall call.Name = "" diff --git a/services/core/internal/sessions/inputs.go b/services/core/internal/sessions/inputs.go index 78706022b..12b31f598 100644 --- a/services/core/internal/sessions/inputs.go +++ b/services/core/internal/sessions/inputs.go @@ -374,6 +374,9 @@ type FunctionCall struct { Arguments json.RawMessage Result json.RawMessage Applied bool + // Subagent and SubagentTurn are the native child and Turn that made the + // call. They are inputs only and are not stored. + Subagent, SubagentTurn string } // FunctionResultInput identifies a persisted call; Result is validated by the API. diff --git a/services/core/tests/integration/function_execution_test.go b/services/core/tests/integration/function_execution_test.go index 031883348..2cf50efe0 100644 --- a/services/core/tests/integration/function_execution_test.go +++ b/services/core/tests/integration/function_execution_test.go @@ -17,17 +17,22 @@ import ( const functionConfiguration = `{"agent":{"model":"gpt-5.5","tools":[{"type":"function","name":"lookup_ticket","description":"Read a synthetic ticket","parameters":{"type":"object","properties":{"ticket":{"type":"string"}},"required":["ticket"],"additionalProperties":false},"defer_loading":false}]},"environment":{"type":"none"}}` func newFunctionHarness(t *testing.T) *dispatchHarness { + t.Helper() + return newFunctionHarnessWith(t, functionConfiguration) +} + +func newFunctionHarnessWith(t *testing.T, configuration string) *dispatchHarness { t.Helper() h := newDispatchHarness(t) var err error - h.session, err = h.s.CreateSession(t.Context(), h.tenant, sessions.CreateSession{Creator: FixtureCreator(), Engine: "codex", IdempotencyKey: "functions", Configuration: json.RawMessage(functionConfiguration)}) + h.session, err = h.s.CreateSession(t.Context(), h.tenant, sessions.CreateSession{Creator: FixtureCreator(), Engine: "codex", IdempotencyKey: "functions", Configuration: json.RawMessage(configuration)}) if err != nil { t.Fatal(err) } if err := bindSessionDevice(t, h.s, h.tenant, h.session.ID, h.device.ID); err != nil { t.Fatal(err) } - h.write("", proto.TypeHeartbeat, proto.HeartbeatPayload{SupportedAgentKinds: []proto.SupportedAgentKind{{Kind: "codex", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{Streaming: proto.CapabilitySupported, Steering: proto.CapabilitySupported, Resume: proto.CapabilitySupported, DurableTurns: proto.CapabilitySupported, DurableInputReceipts: proto.CapabilitySupported, ExecutionControls: proto.CapabilitySupported, SubagentControl: proto.CapabilitySupported, ToolObservations: proto.CapabilitySupported, EnvironmentNone: proto.CapabilitySupported, FunctionTools: proto.CapabilitySupported, FunctionResultImages: proto.CapabilitySupported, Preparation: proto.CapabilitySupported})}}}) + h.write("", proto.TypeHeartbeat, proto.HeartbeatPayload{SupportedAgentKinds: []proto.SupportedAgentKind{{Kind: "codex", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{Streaming: proto.CapabilitySupported, Steering: proto.CapabilitySupported, Resume: proto.CapabilitySupported, DurableTurns: proto.CapabilitySupported, DurableInputReceipts: proto.CapabilitySupported, ExecutionControls: proto.CapabilitySupported, SubagentControl: proto.CapabilitySupported, ToolObservations: proto.CapabilitySupported, EnvironmentNone: proto.CapabilitySupported, FunctionTools: proto.CapabilitySupported, FunctionResultImages: proto.CapabilitySupported, Preparation: proto.CapabilitySupported, SubagentObservations: proto.CapabilitySupported})}}}) deadline := time.Now().Add(time.Second) for { peer, _ := h.registry.LookupDevice(h.device.ID) @@ -116,6 +121,43 @@ func TestExecutionFunctionsWaitForEveryApplicationReceipt(t *testing.T) { h.finished(result, sessions.TurnCompleted) } +func TestExecutionSubagentFunctionCallIsARequiredActionOfTheRootTurn(t *testing.T) { + h := newFunctionHarnessWith(t, strings.Replace(functionConfiguration, `"tools"`, `"id":"agent_root","multi_agent":{"enabled":true},"tools"`, 1)) + input := h.message("start", "Run a child") + result := h.run(t.Context(), input.TurnID) + h.read(testExecutionRequest) + h.write(input.TurnID, proto.TypeSubagentIdentity, proto.SubagentIdentityPayload{NativeID: "child", ParentNativeID: "root", NativeCreatedAt: 100, ParentTurnID: "native-turn", SourceItemID: "spawn-item"}) + h.write(input.TurnID, proto.TypeSubagentTurn, proto.SubagentTurnPayload{NativeID: "child", TurnID: "child-turn", Status: sessions.TurnInProgress, CreatedAtMS: 100000}) + h.write(input.TurnID, proto.TypeSubagentItem, proto.SubagentItemPayload{NativeID: "child", TurnID: "child-turn", ItemID: "a", Position: 0, Kind: proto.TypeToolCall, + Payload: json.RawMessage(`{"id":"a","stage":"before","observation":{"status":"in_progress","kind":"function","name":"lookup_ticket","arguments":{"ticket":"42"}}}`)}) + h.write(input.TurnID, proto.TypeFunctionCall, proto.FunctionCallPayload{CallID: "a", NativeID: "child", TurnID: "child-turn", Name: "lookup_ticket", Arguments: json.RawMessage(`{"ticket":"42"}`)}) + state := functionState(t, h, 1) + child, err := h.s.GetSubagentIdentity(t.Context(), h.tenant, h.session.ID, "child") + if err != nil { + t.Fatal(err) + } + action := state.RequiredActions[0] + childItems, err := sessionAdapter(h.s).ListSubagentItems(t.Context(), h.tenant, h.session.ID, child.ID, "", 20, true) + if err != nil || state.LastTurn.ID != input.TurnID || state.LastTurn.Status != sessions.TurnWaiting || len(childItems.Data) != 1 || childItems.Data[0].CallID != action.CallID { + t.Fatal(state, childItems, err) + } + // The child Turn is not a Session Turn, so only the root Turn answers the call. + answer := json.RawMessage(`{"success":true,"output":"saved"}`) + if err := SubmitFixtureFunctionResult(t.Context(), h.s, h.tenant, h.session.ID, items.Identity(child.ID, "turn:child-turn"), action.CallID, answer); err == nil { + t.Fatal("the child Turn answered the call") + } + if err := SubmitFixtureFunctionResult(t.Context(), h.s, h.tenant, h.session.ID, input.TurnID, action.CallID, answer); err != nil { + t.Fatal(err) + } + var reply proto.FunctionResultPayload + if err := h.read(proto.TypeFunctionResult).DecodePayload(&reply); err != nil || reply.CallID != "a" { + t.Fatal(reply, err) + } + h.write(input.TurnID, proto.TypeInteractionDecisionAck, proto.InteractionDecisionAckPayload{DeliveryID: reply.DeliveryID, Applied: true}) + h.write(input.TurnID, proto.TypeDone, proto.DonePayload{Content: "done"}) + h.finished(result, sessions.TurnCompleted) +} + func TestExecutionFunctionsCancellationAndUnconfirmedResults(t *testing.T) { for _, cancel := range []bool{false, true} { t.Run(map[bool]string{false: "unconfirmed", true: "cancel"}[cancel], func(t *testing.T) { From 7285ed6194f3d6133e446f4bb2fd28dfa0c3d1ea Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 19 Oct 2026 16:10:01 +0000 Subject: [PATCH 2/7] Withdraw function calls the native Harness abandoned --- .../agent/codex/executor_native_test.go | 16 +++- .../internal/agent/codex/executor_turn.go | 11 --- apps/daemon/internal/agent/codex/functions.go | 34 ++++++++ .../internal/agent/codex/functions_test.go | 31 +++++++ .../agent/codex/subagent_snapshots.go | 3 + .../agent/codex/subagent_tools_native_test.go | 22 +---- internal/agentdaemon/proto/functions.go | 14 +++- .../harnessconfig/builtin/admission_test.go | 7 +- .../core/internal/db/queries/functions.sql | 8 +- .../core/internal/db/sqlc/functions.sql.go | 29 +++++-- services/core/internal/db/sqlc/models.go | 1 + services/core/internal/execution/delivery.go | 5 ++ services/core/internal/execution/functions.go | 21 +++++ .../postgres/sessionpg/functions.go | 12 ++- .../internal/sessions/execution_functions.go | 84 +++++++++++++++---- .../sessions/execution_functions_test.go | 40 +++++++++ services/core/internal/sessions/functions.go | 7 +- .../core/internal/sessions/functions_test.go | 6 ++ services/core/internal/sessions/inputs.go | 2 +- services/core/internal/sessions/subagents.go | 13 ++- .../000103_function_call_withdrawn.sql | 9 ++ .../integration/function_execution_test.go | 30 ++++++- 22 files changed, 332 insertions(+), 73 deletions(-) create mode 100644 services/core/migrations/000103_function_call_withdrawn.sql diff --git a/apps/daemon/internal/agent/codex/executor_native_test.go b/apps/daemon/internal/agent/codex/executor_native_test.go index faee82904..e0fdcf6e1 100644 --- a/apps/daemon/internal/agent/codex/executor_native_test.go +++ b/apps/daemon/internal/agent/codex/executor_native_test.go @@ -14,9 +14,12 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" ) -// This opt-in test uses the real pinned harness and an explicitly configured -// Responses provider. Credentials are read from a private file, never arguments. -func TestExecutorNativeReuse(t *testing.T) { +// nativeAcceptanceConfig returns the pinned binary and the explicitly +// configured Responses provider of an opt-in native test, and gives the test an +// isolated Runtime home. Credentials are read from a private file, never +// arguments. +func nativeAcceptanceConfig(t *testing.T) (binary string, options map[string]any) { + t.Helper() if os.Getenv("OAC_TEST_CODEX_EXECUTOR_NATIVE") != "1" { t.Skip("requires an authorized native model acceptance environment") } @@ -45,13 +48,18 @@ func TestExecutorNativeReuse(t *testing.T) { for _, name := range []string{"CODEX_EXEC_SERVER_URL", "CODEX_EXEC_SERVER_NOISE_REGISTRY_URL", "CODEX_EXEC_SERVER_NOISE_ENVIRONMENT_ID", "CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN"} { t.Setenv(name, "") } + return binary, map[string]any{"model": model, "model_provider": map[string]any{"base_url": endpoint, "protocol": "responses", "api_key": strings.TrimSpace(string(key))}} +} + +func TestExecutorNativeReuse(t *testing.T) { + binary, options := nativeAcceptanceConfig(t) cfg := defaultSessionConfig(testStateRoot(t)) cfg.codexBinary = binary cfg.logger = slog.New(slog.DiscardHandler) req := proto.PromptRequestPayload{ AgentKind: "codex", AgentStateKey: "executor-native", DisableExecutionEnvironment: true, DisableSubagents: true, ObserveMessages: true, - AgentOptions: map[string]any{"model": model, "model_provider": map[string]any{"base_url": endpoint, "protocol": "responses", "api_key": strings.TrimSpace(string(key))}}, + AgentOptions: options, FunctionTools: []proto.FunctionTool{{Name: "hold", Description: "Wait until the host supplies a result.", Parameters: json.RawMessage("{\"type\":\"object\",\"properties\":{},\"additionalProperties\":false}")}}, } ctx, cancel := context.WithTimeout(t.Context(), 4*time.Minute) diff --git a/apps/daemon/internal/agent/codex/executor_turn.go b/apps/daemon/internal/agent/codex/executor_turn.go index 747cae1ad..42bc23d87 100644 --- a/apps/daemon/internal/agent/codex/executor_turn.go +++ b/apps/daemon/internal/agent/codex/executor_turn.go @@ -148,19 +148,8 @@ func (s *Session) cancelExecutorTurn(ctx context.Context) error { var _ agent.Turn = (*Session)(nil) -// Server callbacks retain their originating Turn, whose ID must match exactly. func (s *Session) onServerRequest(method string, handler ServerRequestHandler) { s.rpc.OnServerRequest(method, func(raw json.RawMessage, id any) (any, error) { - if s.executor != nil { - var scope struct { - ThreadID string `json:"threadId"` - TurnID string `json:"turnId"` - } - if json.Unmarshal(raw, &scope) != nil || scope.ThreadID == "" || scope.TurnID == "" || - (s.subagents == nil || s.isRootThread(scope.ThreadID)) && !s.isRootTurn(scope.ThreadID, scope.TurnID) { - return nil, errors.New("codex: request does not belong to the active turn") - } - } if !s.beginOperation() { return nil, errors.New("codex: turn has settled") } diff --git a/apps/daemon/internal/agent/codex/functions.go b/apps/daemon/internal/agent/codex/functions.go index b58ba8145..c678d994f 100644 --- a/apps/daemon/internal/agent/codex/functions.go +++ b/apps/daemon/internal/agent/codex/functions.go @@ -132,6 +132,40 @@ func (s *Session) stopFunctionCalls(rootOnly bool) { } } +// withdrawChildCalls removes the pending calls of a child Turn that ended and +// reports them to Core as withdrawn. Native drops their requests without a +// reply, so no result can be applied. A call still held was never reported. +func (s *Session) withdrawChildCalls(ctx context.Context, thread, turn string) error { + if s.functions == nil { + return nil + } + var withdrawn []proto.Envelope + s.functions.mu.Lock() + for id, pending := range s.functions.pending { + if pending.thread != thread || pending.turnID != turn { + continue + } + delete(s.functions.pending, id) + pending.receipt <- agent.ErrUnknownFunctionCall + if pending.held != nil { + continue + } + env, err := proto.NewEnvelope(proto.TypeFunctionCallWithdrawn, s.runID, proto.FunctionCallWithdrawnPayload{CallID: id, NativeID: thread, TurnID: turn}) + if err != nil { + s.functions.mu.Unlock() + return err + } + withdrawn = append(withdrawn, env) + } + s.functions.mu.Unlock() + for _, env := range withdrawn { + if !s.sendWithin(ctx, env) { + return errors.New("codex: function call withdrawal delivery unavailable") + } + } + return nil +} + // sendHeldFunctionCalls sends the Subagent calls whose child Turn Core already // has, because Core resolves a child's call through its identity. func (s *Session) sendHeldFunctionCalls(ctx context.Context) { diff --git a/apps/daemon/internal/agent/codex/functions_test.go b/apps/daemon/internal/agent/codex/functions_test.go index 1c86ee6f2..6019a1680 100644 --- a/apps/daemon/internal/agent/codex/functions_test.go +++ b/apps/daemon/internal/agent/codex/functions_test.go @@ -180,3 +180,34 @@ func TestFunctionToolDefinitionsRejectAmbiguousInput(t *testing.T) { } } } + +func TestEndedChildTurnWithdrawsItsPendingCall(t *testing.T) { + tc, srv, cleanup := NewTestClient() + defer cleanup() + s, out := newFunctionTestSession(tc.JSONRPCClient) + s.subagents = &subagentObservations{wake: make(chan struct{}, 1), sent: map[string]string{proto.TypeSubagentTurn + ":child:child-turn": "{}"}} + s.functions, _ = prepareFunctionTools([]proto.FunctionTool{{Name: "lookup", Parameters: json.RawMessage(`{"type":"object"}`)}}) + s.setThreadID("root") + s.startSteering("root", "turn") + if err := SendServerRequest(srv, "rpc-call", "item/tool/call", map[string]any{"threadId": "child", "turnId": "child-turn", "callId": "call", "tool": "lookup", "arguments": map[string]string{}}); err != nil { + t.Fatal(err) + } + <-s.subagents.wake + s.sendHeldFunctionCalls(t.Context()) + <-out + if err := s.withdrawChildCalls(t.Context(), "child", "child-turn"); err != nil { + t.Fatal(err) + } + var withdrawn proto.FunctionCallWithdrawnPayload + if env := <-out; env.Type != proto.TypeFunctionCallWithdrawn || env.DecodePayload(&withdrawn) != nil || withdrawn != (proto.FunctionCallWithdrawnPayload{CallID: "call", NativeID: "child", TurnID: "child-turn"}) { + t.Fatal(env) + } + if err := s.SubmitFunctionResult(t.Context(), proto.FunctionResultPayload{CallID: "call", Success: true, Content: []proto.InputContent{}}); !errors.Is(err, agent.ErrUnknownFunctionCall) { + t.Fatal(err) + } + select { + case <-s.rpc.Done(): + t.Fatal("a late result closed the native process") + default: + } +} diff --git a/apps/daemon/internal/agent/codex/subagent_snapshots.go b/apps/daemon/internal/agent/codex/subagent_snapshots.go index bbe4310d1..19bd8c3fe 100644 --- a/apps/daemon/internal/agent/codex/subagent_snapshots.go +++ b/apps/daemon/internal/agent/codex/subagent_snapshots.go @@ -218,6 +218,9 @@ func (s *Session) publishSubagentHistory(ctx context.Context, h subagentHistory, if err := s.sendSubagentFact(ctx, proto.TypeSubagentTurn, key, payload); err != nil { return false, err } + if err := s.withdrawChildCalls(ctx, h.ID, turn.ID); err != nil { + return false, err + } } } return busy || nativeStatus.Type == "active", nil diff --git a/apps/daemon/internal/agent/codex/subagent_tools_native_test.go b/apps/daemon/internal/agent/codex/subagent_tools_native_test.go index bb5e37353..4d0840434 100644 --- a/apps/daemon/internal/agent/codex/subagent_tools_native_test.go +++ b/apps/daemon/internal/agent/codex/subagent_tools_native_test.go @@ -7,7 +7,6 @@ import ( "log/slog" "net/http" "net/http/httptest" - "os" "strings" "sync" "testing" @@ -20,21 +19,7 @@ import ( // This opt-in test runs the real pinned harness with a model that spawns a // Subagent, which calls a function or an MCP tool of the Agent. func TestSubagentToolsNative(t *testing.T) { - if os.Getenv("OAC_TEST_CODEX_EXECUTOR_NATIVE") != "1" { - t.Skip("requires an authorized native model acceptance environment") - } - binary, root := os.Getenv("OAC_TEST_CODEX_BINARY"), os.Getenv("OAC_TEST_CODEX_LIVE_ROOT") - model, endpoint := os.Getenv("OAC_TEST_CODEX_MODEL"), os.Getenv("OAC_TEST_CODEX_BASE_URL") - key, err := os.ReadFile(os.Getenv("OAC_TEST_CODEX_KEY_FILE")) - if binary == "" || root == "" || model == "" || endpoint == "" || err != nil { - t.Fatal("missing explicit native acceptance configuration") - } - isolated, err := os.MkdirTemp(root, "run-") - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = os.RemoveAll(isolated) }) - t.Setenv("OAC_RUNTIME_HOME", isolated) + binary, options := nativeAcceptanceConfig(t) mcpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, _ := io.ReadAll(r.Body) var rpc struct { @@ -70,7 +55,7 @@ func TestSubagentToolsNative(t *testing.T) { req := proto.PromptRequestPayload{ AgentKind: "codex", AgentStateKey: "subagent-tools-" + name, DisableExecutionEnvironment: true, ObserveSubagentIdentities: true, - AgentOptions: map[string]any{"model": model, "model_provider": map[string]any{"base_url": endpoint, "protocol": "responses", "api_key": strings.TrimSpace(string(key))}}, + AgentOptions: options, FunctionTools: []proto.FunctionTool{{Name: "lookup", Description: "Return the synthetic verification value.", Parameters: json.RawMessage(`{"type":"object","properties":{},"additionalProperties":false}`)}}, MCPHTTPServers: &servers, } @@ -86,7 +71,6 @@ func TestSubagentToolsNative(t *testing.T) { if err != nil { t.Fatal(err) } - var mu sync.Mutex var calls []proto.FunctionCallPayload var childTools []string var answer string @@ -96,9 +80,7 @@ func TestSubagentToolsNative(t *testing.T) { case proto.TypeFunctionCall: var call proto.FunctionCallPayload _ = event.DecodePayload(&call) - mu.Lock() calls = append(calls, call) - mu.Unlock() text := "FN-4729" answers.Add(1) go func() { diff --git a/internal/agentdaemon/proto/functions.go b/internal/agentdaemon/proto/functions.go index d9acd9764..0ec33b5d0 100644 --- a/internal/agentdaemon/proto/functions.go +++ b/internal/agentdaemon/proto/functions.go @@ -6,8 +6,9 @@ import ( ) const ( - TypeFunctionCall = "function_call" - TypeFunctionResult = "function_result" + TypeFunctionCall = "function_call" + TypeFunctionCallWithdrawn = "function_call_withdrawn" + TypeFunctionResult = "function_result" ) type FunctionTool struct { @@ -48,6 +49,15 @@ type FunctionCallPayload struct { Arguments json.RawMessage `json:"arguments"` } +// FunctionCallWithdrawnPayload tells that the native Harness will never apply a +// result for the call, which CallID, NativeID and TurnID identify as in +// FunctionCallPayload. +type FunctionCallWithdrawnPayload struct { + CallID string `json:"call_id"` + NativeID string `json:"native_id,omitempty"` + TurnID string `json:"turn_id,omitempty"` +} + type FunctionResultPayload struct { DeliveryID string `json:"delivery_id"` CallID string `json:"call_id"` diff --git a/internal/harnessconfig/builtin/admission_test.go b/internal/harnessconfig/builtin/admission_test.go index 00b46c53d..47c2698f2 100644 --- a/internal/harnessconfig/builtin/admission_test.go +++ b/internal/harnessconfig/builtin/admission_test.go @@ -39,12 +39,9 @@ func TestMiniMaxMCPPoliciesRejectInsteadOfDropping(t *testing.T) { } func TestSubagentDeclarationsPreserveSupportedBoundaries(t *testing.T) { - for _, kind := range []string{"claude_sdk", "codex", "mcode"} { + for _, kind := range []string{"claude_sdk", "codex"} { t.Run(kind, func(t *testing.T) { declaration, _ := Registry().Lookup(kind) - if declaration.ValidateConfiguration == nil { - return // Only the common checks apply. - } admission := harnessconfig.Admission{MultiAgent: true} for _, placement := range []string{"none", "openai_hosted", "self_hosted"} { admission.Placement = placement @@ -52,7 +49,7 @@ func TestSubagentDeclarationsPreserveSupportedBoundaries(t *testing.T) { t.Fatal(placement, err) } } - // Codex and MiniMax Code admit function and MCP tools with Subagents. + // Codex admits function and MCP tools with Subagents; Claude Code does not. admission.Placement = "none" for _, tool := range []string{`{"type":"function","name":"f"}`, `{"type":"mcp","server_label":"s"}`} { admission.Tools = []json.RawMessage{json.RawMessage(tool)} diff --git a/services/core/internal/db/queries/functions.sql b/services/core/internal/db/queries/functions.sql index 81a78b857..783d67234 100644 --- a/services/core/internal/db/queries/functions.sql +++ b/services/core/internal/db/queries/functions.sql @@ -14,7 +14,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 +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 ORDER BY f.created_at, f.call_id; @@ -24,7 +24,8 @@ SELECT EXISTS ( )::boolean AS found; -- name: MatchFunctionResult :one -SELECT (result IS NOT NULL)::boolean AS submitted, COALESCE(result = sqlc.arg(result)::jsonb, false)::boolean AS matches +SELECT (result IS NOT NULL)::boolean AS submitted, COALESCE(result = sqlc.arg(result)::jsonb, false)::boolean AS matches, + withdrawn FROM function_calls WHERE session_id = sqlc.arg(session_id) AND turn_id = sqlc.arg(turn_id) AND call_id = sqlc.arg(call_id); @@ -34,6 +35,9 @@ UPDATE function_calls SET result = $4 WHERE session_id = $1 AND turn_id = $2 AND -- name: ApplyFunctionResult :exec UPDATE function_calls SET applied = true WHERE session_id = $1 AND turn_id = $2 AND call_id = $3; +-- name: WithdrawFunctionCall :exec +UPDATE function_calls SET withdrawn = true WHERE session_id = $1 AND turn_id = $2 AND call_id = $3 AND NOT applied; + -- name: FunctionItemResult :one SELECT result FROM function_calls WHERE session_id = $1 AND turn_id = $2 AND call_id = $3; diff --git a/services/core/internal/db/sqlc/functions.sql.go b/services/core/internal/db/sqlc/functions.sql.go index 4594004cb..887a84cf7 100644 --- a/services/core/internal/db/sqlc/functions.sql.go +++ b/services/core/internal/db/sqlc/functions.sql.go @@ -74,7 +74,7 @@ func (q *Queries) FunctionItemResult(ctx context.Context, arg FunctionItemResult } const getFunctionCall = `-- name: GetFunctionCall :one -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 FROM function_calls f JOIN sessions s ON s.id = f.session_id +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 sessions s ON s.id = f.session_id WHERE s.tenant_id = $1 AND f.session_id = $2 AND f.turn_id = $3 AND f.call_id = $4 ` @@ -103,13 +103,14 @@ func (q *Queries) GetFunctionCall(ctx context.Context, arg GetFunctionCallParams &i.Result, &i.Applied, &i.CreatedAt, + &i.Withdrawn, ) return i, err } 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 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 +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 ORDER BY f.created_at, f.call_id ` @@ -138,6 +139,7 @@ func (q *Queries) ListPendingFunctionCalls(ctx context.Context, arg ListPendingF &i.Result, &i.Applied, &i.CreatedAt, + &i.Withdrawn, ); err != nil { return nil, err } @@ -180,7 +182,8 @@ func (q *Queries) MatchFunctionCall(ctx context.Context, arg MatchFunctionCallPa } const matchFunctionResult = `-- name: MatchFunctionResult :one -SELECT (result IS NOT NULL)::boolean AS submitted, COALESCE(result = $1::jsonb, false)::boolean AS matches +SELECT (result IS NOT NULL)::boolean AS submitted, COALESCE(result = $1::jsonb, false)::boolean AS matches, + withdrawn FROM function_calls WHERE session_id = $2 AND turn_id = $3 AND call_id = $4 ` @@ -195,6 +198,7 @@ type MatchFunctionResultParams struct { type MatchFunctionResultRow struct { Submitted bool `json:"submitted"` Matches bool `json:"matches"` + Withdrawn bool `json:"withdrawn"` } func (q *Queries) MatchFunctionResult(ctx context.Context, arg MatchFunctionResultParams) (MatchFunctionResultRow, error) { @@ -205,7 +209,7 @@ func (q *Queries) MatchFunctionResult(ctx context.Context, arg MatchFunctionResu arg.CallID, ) var i MatchFunctionResultRow - err := row.Scan(&i.Submitted, &i.Matches) + err := row.Scan(&i.Submitted, &i.Matches, &i.Withdrawn) return i, err } @@ -247,3 +251,18 @@ func (q *Queries) SubmitFunctionResult(ctx context.Context, arg SubmitFunctionRe ) return err } + +const withdrawFunctionCall = `-- name: WithdrawFunctionCall :exec +UPDATE function_calls SET withdrawn = true WHERE session_id = $1 AND turn_id = $2 AND call_id = $3 AND NOT applied +` + +type WithdrawFunctionCallParams struct { + SessionID pgtype.UUID `json:"session_id"` + TurnID pgtype.UUID `json:"turn_id"` + CallID string `json:"call_id"` +} + +func (q *Queries) WithdrawFunctionCall(ctx context.Context, arg WithdrawFunctionCallParams) error { + _, err := q.db.Exec(ctx, withdrawFunctionCall, arg.SessionID, arg.TurnID, arg.CallID) + return err +} diff --git a/services/core/internal/db/sqlc/models.go b/services/core/internal/db/sqlc/models.go index 178b78211..4a12508fd 100644 --- a/services/core/internal/db/sqlc/models.go +++ b/services/core/internal/db/sqlc/models.go @@ -170,6 +170,7 @@ type FunctionCall struct { Result []byte `json:"result"` Applied bool `json:"applied"` CreatedAt pgtype.Timestamptz `json:"created_at"` + Withdrawn bool `json:"withdrawn"` } type InitialEnvironmentFile struct { diff --git a/services/core/internal/execution/delivery.go b/services/core/internal/execution/delivery.go index df5b00716..1ec0e631b 100644 --- a/services/core/internal/execution/delivery.go +++ b/services/core/internal/execution/delivery.go @@ -226,6 +226,11 @@ func (d *Dispatcher) deliver(ctx context.Context, tenantID, sessionID string, pe result.ErrorCode = "function_call_invalid" return } + case proto.TypeFunctionCallWithdrawn: + if err := functions.withdraw(ctx, env); err != nil { + result.ErrorCode = "function_call_invalid" + return + } case proto.TypePromptSteerAck: var ack proto.PromptSteerAckPayload if env.DecodePayload(&ack) != nil { diff --git a/services/core/internal/execution/functions.go b/services/core/internal/execution/functions.go index 39c1d9bf3..96f7d4b2d 100644 --- a/services/core/internal/execution/functions.go +++ b/services/core/internal/execution/functions.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strings" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" @@ -74,6 +75,18 @@ func (f *functionExchange) record(ctx context.Context, env proto.Envelope) error return f.unlessCancelling(ctx, err) } +func (f *functionExchange) withdraw(ctx context.Context, env proto.Envelope) error { + var call proto.FunctionCallWithdrawnPayload + if env.ID != f.turn || env.DecodePayload(&call) != nil { + return errors.New("invalid function withdrawal") + } + err := f.sessions.WithdrawFunctionCall(ctx, f.tenant, f.session, f.turn, sessions.FunctionCall{ + CallID: items.Identity(f.turn, "tool:"+call.CallID), ExecutorCallID: call.CallID, + Subagent: call.NativeID, SubagentTurn: call.TurnID, + }) + return f.unlessCancelling(ctx, err) +} + func (f *functionExchange) start(ctx context.Context, peer *runtimegateway.Session) error { if f.reply != nil || len(f.tools) == 0 { return nil @@ -115,6 +128,14 @@ func (f *functionExchange) confirm(ctx context.Context, reply functionReply) err return reply.err } if !reply.ack.Applied { + // A result in flight when the native Harness abandoned the call is moot. + calls, err := f.sessions.PendingFunctionCalls(ctx, f.tenant, f.session, f.turn) + if err != nil { + return err + } + if !slices.ContainsFunc(calls, func(call sessions.FunctionCall) bool { return call.CallID == id }) { + return nil + } return fmt.Errorf("function result not applied: %s", reply.ack.ErrorCode) } return f.unlessCancelling(ctx, f.sessions.ConfirmFunctionResult(ctx, f.tenant, f.session, f.turn, id)) diff --git a/services/core/internal/persistence/postgres/sessionpg/functions.go b/services/core/internal/persistence/postgres/sessionpg/functions.go index 8f82b4fdb..3ee5cb2d8 100644 --- a/services/core/internal/persistence/postgres/sessionpg/functions.go +++ b/services/core/internal/persistence/postgres/sessionpg/functions.go @@ -53,7 +53,7 @@ func loadPendingFunctionCalls(ctx context.Context, q *sqlc.Queries, session pgty } func functionCallFromRow(row sqlc.FunctionCall) sessions.FunctionCall { - return sessions.FunctionCall{CallID: row.CallID, ExecutorCallID: row.ExecutorCallID, Name: row.Name, Arguments: row.Arguments, Result: row.Result, Applied: row.Applied} + return sessions.FunctionCall{CallID: row.CallID, ExecutorCallID: row.ExecutorCallID, Name: row.Name, Arguments: row.Arguments, Result: row.Result, Applied: row.Applied, Withdrawn: row.Withdrawn} } func (t *SessionTx) LoadPendingFunctionCalls(ctx context.Context, turn string) ([]sessions.FunctionCall, error) { @@ -83,7 +83,7 @@ func (t *SessionTx) MatchFunctionResult(ctx context.Context, turn, call string, if err != nil { return sessions.FunctionResultMatch{}, err } - return sessions.FunctionResultMatch{Recorded: true, Submitted: row.Submitted, Matches: row.Matches}, nil + return sessions.FunctionResultMatch{Recorded: true, Submitted: row.Submitted, Matches: row.Matches, Withdrawn: row.Withdrawn}, nil } func (t *SessionTx) SubmitFunctionResult(ctx context.Context, turn, call string, result json.RawMessage) error { @@ -150,3 +150,11 @@ func (t *SessionTx) ApplyFunctionResult(ctx context.Context, turn, call string) } return t.q.ApplyFunctionResult(ctx, sqlc.ApplyFunctionResultParams{SessionID: t.session, TurnID: id, CallID: call}) } + +func (t *SessionTx) WithdrawFunctionCall(ctx context.Context, turn, call string) error { + id, err := parseID(turn) + if err != nil { + return err + } + return t.q.WithdrawFunctionCall(ctx, sqlc.WithdrawFunctionCallParams{SessionID: t.session, TurnID: id, CallID: call}) +} diff --git a/services/core/internal/sessions/execution_functions.go b/services/core/internal/sessions/execution_functions.go index a39d49e6d..9495cc8fe 100644 --- a/services/core/internal/sessions/execution_functions.go +++ b/services/core/internal/sessions/execution_functions.go @@ -22,9 +22,7 @@ type FunctionCallMatch struct { type FunctionTx interface { JournalTx PendingFunctionCallsTx - // LoadNativeSubagent reads the Session's native Subagent and reports - // whether the Session has it. - LoadNativeSubagent(ctx context.Context, native string) (NativeSubagent, bool, error) + NativeSubagents // MatchFunctionCall compares call with the Turn's call of the same public // identity. MatchFunctionCall(ctx context.Context, turn string, call FunctionCall) (FunctionCallMatch, error) @@ -36,6 +34,8 @@ type FunctionTx interface { // ApplyFunctionResult records the application receipt of the call's // result. ApplyFunctionResult(ctx context.Context, turn, call string) error + // WithdrawFunctionCall records that the native Harness abandoned the call. + WithdrawFunctionCall(ctx context.Context, turn, call string) error // ApplyTurnStatus moves the Turn as change decides and returns it as // stored. A Turn no longer in the expected status is ErrTurnConflict. ApplyTurnStatus(ctx context.Context, turn string, change TurnStatusChange) (Turn, error) @@ -69,17 +69,49 @@ func decideFunctionCall(turn Turn, match FunctionCallMatch) (bool, error) { // decideFunctionConfirmation decides whether to record the native application // of a call's result: an applied call records nothing, and a call without a -// result or on a Turn that no longer accepts results conflicts. +// result, a withdrawn call or a call on a Turn that no longer accepts results +// conflicts. func decideFunctionConfirmation(turn Turn, call FunctionCall) (bool, error) { if call.Applied { return false, nil } - if len(call.Result) == 0 || !acceptsFunctionResults(turn) { + if len(call.Result) == 0 || call.Withdrawn || !acceptsFunctionResults(turn) { return false, ErrTurnConflict } return true, nil } +// decideFunctionWithdrawal decides whether to record that the native Harness +// abandoned a call: a withdrawn call records nothing, and an applied call or a +// Turn that no longer accepts results conflicts. +func decideFunctionWithdrawal(turn Turn, call FunctionCall) (bool, error) { + if call.Withdrawn { + return false, nil + } + if call.Applied || !acceptsFunctionResults(turn) { + return false, ErrTurnConflict + } + return true, nil +} + +// resolveFunctionCall gives a call a Subagent made the identity of the child's +// function Item. A child the Session has not published, or a malformed native +// identity, is ErrInvalidInput. +func resolveFunctionCall(ctx context.Context, tx NativeSubagents, call FunctionCall) (FunctionCall, error) { + if call.Subagent == "" && call.SubagentTurn == "" { + return call, nil + } + if !validNativeIdentity(call.Subagent) || !validNativeIdentity(call.SubagentTurn) { + return call, ErrInvalidInput + } + child, err := visibleSubagent(ctx, tx, call.Subagent) + if err != nil { + return call, err + } + call.CallID = items.Identity(items.Identity(child.ID, "turn:"+call.SubagentTurn), "tool:"+call.ExecutorCallID) + return call, nil +} + // RecordFunctionCall records a call an executor reported for its Turn and // journals the Turn's required actions with it. A call a Subagent made takes // the identity of the child's function Item. A retry of the recorded call @@ -89,15 +121,9 @@ func (o *ExecutionOperations) RecordFunctionCall(ctx context.Context, tenant, se return err } return o.storage.WithFunctionTurn(ctx, tenant, session, turn, func(ctx context.Context, tx FunctionTx, current Turn) error { - if call.Subagent != "" || call.SubagentTurn != "" { - child, found, err := tx.LoadNativeSubagent(ctx, call.Subagent) - if err != nil { - return err - } - if !found || !child.Visible || !validNativeIdentity(call.SubagentTurn) { - return ErrInvalidInput - } - call.CallID = items.Identity(items.Identity(child.ID, "turn:"+call.SubagentTurn), "tool:"+call.ExecutorCallID) + call, err := resolveFunctionCall(ctx, tx, call) + if err != nil { + return err } match, err := tx.MatchFunctionCall(ctx, current.ID, call) if err != nil { @@ -155,6 +181,36 @@ func (o *ExecutionOperations) ConfirmFunctionResult(ctx context.Context, tenant, }) } +// WithdrawFunctionCall records that the native Harness will never apply a +// result for the call, which stops being a required action, and journals the +// Turn's remaining required actions. A repeated withdrawal changes nothing. +func (o *ExecutionOperations) WithdrawFunctionCall(ctx context.Context, tenant, session, turn string, call FunctionCall) error { + if !validFunctionIdentity(call.CallID) || !validFunctionIdentity(call.ExecutorCallID) { + return ErrInvalidInput + } + return o.storage.WithFunctionTurn(ctx, tenant, session, turn, func(ctx context.Context, tx FunctionTx, current Turn) error { + call, err := resolveFunctionCall(ctx, tx, call) + if err != nil { + return err + } + stored, found, err := tx.LoadFunctionCall(ctx, current.ID, call.CallID) + if err != nil { + return err + } + if !found { + return ErrNotFound + } + withdraw, err := decideFunctionWithdrawal(current, stored) + if err != nil || !withdraw { + return err + } + if err := tx.WithdrawFunctionCall(ctx, current.ID, call.CallID); err != nil { + return err + } + return settleFunctionState(ctx, tx, current) + }) +} + // settleFunctionState moves a Turn that accepts results to waiting while it // has required actions and back to in progress once it has none, journals // that transition, then journals the Session activity with the required diff --git a/services/core/internal/sessions/execution_functions_test.go b/services/core/internal/sessions/execution_functions_test.go index 62deff254..811e9c59b 100644 --- a/services/core/internal/sessions/execution_functions_test.go +++ b/services/core/internal/sessions/execution_functions_test.go @@ -285,3 +285,43 @@ func TestPendingFunctionCalls(t *testing.T) { t.Fatal(calls, err) } } + +func TestWithdrawFunctionCall(t *testing.T) { + stored := FunctionCall{CallID: "call"} + t.Run("last withdrawal resumes the Turn", func(t *testing.T) { + f := &fakeFunctionTx{fakeTx: &fakeTx{t: t, loadUsage: returns(json.RawMessage(`{}`)), appendChanges: func([]SessionChange) error { return nil }, + applyTurnStatus: func(TurnStatusChange) (Turn, error) { return turnWith(TurnInProgress), nil }}, + loadFunctionCall: func() (FunctionCall, bool, error) { return stored, true, nil }, withdrawFunctionCall: done, + loadPendingFunctionCalls: returns([]FunctionCall{})} + if err := functionOperations(t, f, turnWith(TurnWaiting)).WithdrawFunctionCall(t.Context(), testTenant, testSession, testTurn, testCall); err != nil { + t.Fatal(err) + } + assertCalls(t, f.fakeTx, "LoadFunctionCall "+testTurn+" call", "WithdrawFunctionCall "+testTurn+" call", "LoadPendingFunctionCalls "+testTurn, + "ApplyTurnStatus "+testTurn+" waiting in_progress {}", "AppendChanges agent.session.turn.in_progress", "LoadUsage", "AppendChanges agent.session.in_progress") + }) + t.Run("repeated withdrawal records nothing", func(t *testing.T) { + withdrawn := stored + withdrawn.Withdrawn = true + f := &fakeFunctionTx{fakeTx: &fakeTx{t: t}, loadFunctionCall: func() (FunctionCall, bool, error) { return withdrawn, true, nil }} + if err := functionOperations(t, f, turnWith(TurnInProgress)).WithdrawFunctionCall(t.Context(), testTenant, testSession, testTurn, testCall); err != nil { + t.Fatal(err) + } + assertCalls(t, f.fakeTx, "LoadFunctionCall "+testTurn+" call") + }) + t.Run("an applied call conflicts", func(t *testing.T) { + applied := stored + applied.Applied = true + f := &fakeFunctionTx{fakeTx: &fakeTx{t: t}, loadFunctionCall: func() (FunctionCall, bool, error) { return applied, true, nil }} + if err := functionOperations(t, f, turnWith(TurnWaiting)).WithdrawFunctionCall(t.Context(), testTenant, testSession, testTurn, testCall); !errors.Is(err, ErrTurnConflict) { + t.Fatal(err) + } + }) + t.Run("a malformed Subagent identity is invalid input", func(t *testing.T) { + call := testCall + call.Subagent, call.SubagentTurn = "child\n", "child-turn" + f := &fakeFunctionTx{fakeTx: &fakeTx{t: t}} + if err := functionOperations(t, f, turnWith(TurnWaiting)).WithdrawFunctionCall(t.Context(), testTenant, testSession, testTurn, call); !errors.Is(err, ErrInvalidInput) { + t.Fatal(err) + } + }) +} diff --git a/services/core/internal/sessions/functions.go b/services/core/internal/sessions/functions.go index df4832f21..7d8008bdd 100644 --- a/services/core/internal/sessions/functions.go +++ b/services/core/internal/sessions/functions.go @@ -60,13 +60,18 @@ type FunctionResultMatch struct { Submitted bool // Matches reports that the saved result equals the submitted one. Matches bool + // Withdrawn reports that the native Harness abandoned the call. + Withdrawn bool } // decideFunctionResult decides whether to store a submitted result for a call -// the Turn has. An identical retry of a saved result stores nothing, even after +// the Turn has. A withdrawn call takes no result. An identical retry of a saved result stores nothing, even after // the Turn ended; a different result conflicts with the saved one. A new result // needs a Turn that accepts results. func decideFunctionResult(turn Turn, match FunctionResultMatch) (bool, error) { + if match.Withdrawn { + return false, ErrTurnConflict + } if match.Submitted { if !match.Matches { return false, ErrFunctionResultConflict diff --git a/services/core/internal/sessions/functions_test.go b/services/core/internal/sessions/functions_test.go index 76d661868..71600da65 100644 --- a/services/core/internal/sessions/functions_test.go +++ b/services/core/internal/sessions/functions_test.go @@ -24,6 +24,7 @@ type fakeFunctionTx struct { createFunctionCall func() error loadFunctionCall func() (FunctionCall, bool, error) applyFunctionResult func() error + withdrawFunctionCall func() error } var ( @@ -76,6 +77,11 @@ func (f *fakeFunctionTx) ApplyFunctionResult(_ context.Context, turn, call strin return f.applyFunctionResult() } +func (f *fakeFunctionTx) WithdrawFunctionCall(_ context.Context, turn, call string) error { + f.record("WithdrawFunctionCall", f.withdrawFunctionCall != nil, turn, call) + return f.withdrawFunctionCall() +} + func cancelling(status string) Turn { turn := turnWith(status) turn.CancelRequestedAt = time.Unix(1700000001, 0) diff --git a/services/core/internal/sessions/inputs.go b/services/core/internal/sessions/inputs.go index 12b31f598..c2159017d 100644 --- a/services/core/internal/sessions/inputs.go +++ b/services/core/internal/sessions/inputs.go @@ -373,7 +373,7 @@ type FunctionCall struct { CallID, ExecutorCallID, Name string Arguments json.RawMessage Result json.RawMessage - Applied bool + Applied, Withdrawn bool // Subagent and SubagentTurn are the native child and Turn that made the // call. They are inputs only and are not stored. Subagent, SubagentTurn string diff --git a/services/core/internal/sessions/subagents.go b/services/core/internal/sessions/subagents.go index 14327ff31..458263d9e 100644 --- a/services/core/internal/sessions/subagents.go +++ b/services/core/internal/sessions/subagents.go @@ -92,14 +92,19 @@ type ChildItem struct { Output bool } +// NativeSubagents reads the Session's native Subagents. +type NativeSubagents interface { + // LoadNativeSubagent reads the Session's native Subagent and reports whether + // the Session has it. + LoadNativeSubagent(ctx context.Context, native string) (NativeSubagent, bool, error) +} + // SubagentProjectionTx loads and applies what projecting Subagent observations // reads and writes. type SubagentProjectionTx interface { + NativeSubagents // LoadRootAgent reads the Session's Agent ID. LoadRootAgent(ctx context.Context) (string, error) - // LoadNativeSubagent reads the Session's native Subagent and reports whether - // the Session has it. - LoadNativeSubagent(ctx context.Context, native string) (NativeSubagent, bool, error) // PutSubagentIdentity binds a native Subagent to the Session, or finds the // same binding, and returns its ID. A different binding of the native ID, // or a parent the Session does not know, is ErrIdempotencyConflict. @@ -138,7 +143,7 @@ func validNativeIdentity(value string) bool { // visibleSubagent reads the public native Subagent an observation names; one // the Session does not have or has not published is ErrInvalidInput. -func visibleSubagent(ctx context.Context, tx SubagentProjectionTx, native string) (NativeSubagent, error) { +func visibleSubagent(ctx context.Context, tx NativeSubagents, native string) (NativeSubagent, error) { child, found, err := tx.LoadNativeSubagent(ctx, native) if err != nil { return NativeSubagent{}, err diff --git a/services/core/migrations/000103_function_call_withdrawn.sql b/services/core/migrations/000103_function_call_withdrawn.sql new file mode 100644 index 000000000..861c7028b --- /dev/null +++ b/services/core/migrations/000103_function_call_withdrawn.sql @@ -0,0 +1,9 @@ +-- +goose Up +-- A native Harness can abandon a call before it applies a result. A withdrawn call is no longer a required action. +ALTER TABLE function_calls ADD COLUMN withdrawn boolean NOT NULL DEFAULT false; +ALTER TABLE function_calls ADD CONSTRAINT function_calls_not_applied_and_withdrawn CHECK (NOT (applied AND withdrawn)); +DROP INDEX function_calls_pending_idx; +CREATE INDEX function_calls_pending_idx ON function_calls(session_id, turn_id) WHERE NOT applied AND NOT withdrawn; + +-- +goose Down +SELECT 1; diff --git a/services/core/tests/integration/function_execution_test.go b/services/core/tests/integration/function_execution_test.go index 2cf50efe0..c9aff3196 100644 --- a/services/core/tests/integration/function_execution_test.go +++ b/services/core/tests/integration/function_execution_test.go @@ -143,8 +143,8 @@ func TestExecutionSubagentFunctionCallIsARequiredActionOfTheRootTurn(t *testing. } // The child Turn is not a Session Turn, so only the root Turn answers the call. answer := json.RawMessage(`{"success":true,"output":"saved"}`) - if err := SubmitFixtureFunctionResult(t.Context(), h.s, h.tenant, h.session.ID, items.Identity(child.ID, "turn:child-turn"), action.CallID, answer); err == nil { - t.Fatal("the child Turn answered the call") + if err := SubmitFixtureFunctionResult(t.Context(), h.s, h.tenant, h.session.ID, items.Identity(child.ID, "turn:child-turn"), action.CallID, answer); !errors.Is(err, sessions.ErrFunctionCallTurnMismatch) { + t.Fatal("the child Turn answered the call", err) } if err := SubmitFixtureFunctionResult(t.Context(), h.s, h.tenant, h.session.ID, input.TurnID, action.CallID, answer); err != nil { t.Fatal(err) @@ -158,6 +158,32 @@ func TestExecutionSubagentFunctionCallIsARequiredActionOfTheRootTurn(t *testing. h.finished(result, sessions.TurnCompleted) } +func TestExecutionWithdrawnFunctionCallStopsBeingARequiredAction(t *testing.T) { + h := newFunctionHarnessWith(t, strings.Replace(functionConfiguration, `"tools"`, `"id":"agent_root","multi_agent":{"enabled":true},"tools"`, 1)) + input := h.message("start", "Run a child") + result := h.run(t.Context(), input.TurnID) + h.read(testExecutionRequest) + h.write(input.TurnID, proto.TypeSubagentIdentity, proto.SubagentIdentityPayload{NativeID: "child", ParentNativeID: "root", NativeCreatedAt: 100, ParentTurnID: "native-turn", SourceItemID: "spawn-item"}) + h.write(input.TurnID, proto.TypeSubagentTurn, proto.SubagentTurnPayload{NativeID: "child", TurnID: "child-turn", Status: sessions.TurnInProgress, CreatedAtMS: 100000}) + call := proto.FunctionCallPayload{CallID: "a", NativeID: "child", TurnID: "child-turn", Name: "lookup_ticket", Arguments: json.RawMessage(`{"ticket":"42"}`)} + h.write(input.TurnID, proto.TypeFunctionCall, call) + action := functionState(t, h, 1).RequiredActions[0] + withdrawn := proto.FunctionCallWithdrawnPayload{CallID: call.CallID, NativeID: call.NativeID, TurnID: call.TurnID} + for range 2 { + h.write(input.TurnID, proto.TypeFunctionCallWithdrawn, withdrawn) + } + state := functionState(t, h, 0) + if state.LastTurn.Status != sessions.TurnInProgress { + t.Fatal(state.LastTurn) + } + answer := json.RawMessage(`{"success":true,"output":"late"}`) + if err := SubmitFixtureFunctionResult(t.Context(), h.s, h.tenant, h.session.ID, input.TurnID, action.CallID, answer); !errors.Is(err, sessions.ErrTurnConflict) { + t.Fatal(err) + } + h.write(input.TurnID, proto.TypeDone, proto.DonePayload{Content: "done"}) + h.finished(result, sessions.TurnCompleted) +} + func TestExecutionFunctionsCancellationAndUnconfirmedResults(t *testing.T) { for _, cancel := range []bool{false, true} { t.Run(map[bool]string{false: "unconfirmed", true: "cancel"}[cancel], func(t *testing.T) { From f7637fdac12d02e3306bb1fb0c44ef6e0bfefe0b Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 19 Oct 2026 16:11:06 +0000 Subject: [PATCH 3/7] Document withdrawn function calls and fix review findings --- contracts/agents-api/execution-tools.md | 3 +++ contracts/agents-api/index.md | 3 ++- contracts/agents-api/message-content.md | 4 ++-- contracts/agents-api/subagents.md | 2 +- contracts/agents-api/zh/execution-tools.md | 5 ++++- contracts/agents-api/zh/harness-capabilities.md | 4 ++-- contracts/agents-api/zh/index.md | 5 +++-- contracts/agents-api/zh/message-content.md | 6 +++--- contracts/agents-api/zh/subagents.md | 4 ++-- docs/runtime-protocol.md | 4 +++- docs/zh/runtime-protocol.md | 6 ++++-- 11 files changed, 29 insertions(+), 17 deletions(-) diff --git a/contracts/agents-api/execution-tools.md b/contracts/agents-api/execution-tools.md index e69ec24e0..b82f1e655 100644 --- a/contracts/agents-api/execution-tools.md +++ b/contracts/agents-api/execution-tools.md @@ -22,11 +22,14 @@ A function declaration requires `name`, `description` and a JSON Schema in `para | Identical retry, also after the Turn ends | Accepted | | Different result for the same call | 409 `conflict_error` | | First result after the Turn was cancelled | 409 `conflict_error` | +| Result for a call the native Harness abandoned | 409 `conflict_error` | | Unknown call, or a call of another Turn, in the caller's Session | 400 `invalid_request_error`; the pending action is unchanged | | Missing or foreign Session | 404 | [Session input conflicts](./sessions-events.md#input-errors) records the exact messages. Invalid or unsupported content cannot consume a pending call. Admission is separate from application: the adapter confirms a result only when the matching native tool result appears in the live Turn that made the call ([receipt contract](./message-content.md#function-results)). A transport write alone confirms nothing, and a confirmation says nothing about provider consumption or exactly-once external effects. Core never replays a result automatically. +The native Harness can abandon a call before it applies a result, for example when the root Turn closes or interrupts the Subagent that made it. The call then stops being a required action, the Turn continues and returns to `in_progress` when no required action is left, and a later `function_call_output` for it is rejected like a result after cancellation. An abandoned call has no application receipt. + A function call made by a Subagent is a required action of the root Turn that runs the child. Its `turn_id` is that root Turn, and its `call_id` equals the `call_id` of the child's `function_call` Item on the [Subagent routes](./subagents.md#public-reads). Answer it with `function_call_output` / `tool_result` under that `call_id` and the root Turn ID; the child's Turn ID is not a Session Turn and gets the 400 for a call of another Turn. A Subagent's HTTP MCP calls are `mcp_call` Items in the child's history. ### Required actions and recovery diff --git a/contracts/agents-api/index.md b/contracts/agents-api/index.md index d82944228..b189df30f 100644 --- a/contracts/agents-api/index.md +++ b/contracts/agents-api/index.md @@ -81,6 +81,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. - 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 call the native Harness abandons, such as a Codex Subagent call dropped when the root closes or interrupts the child, 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`. - Error messages omit the call and executor IDs that official messages include. @@ -151,7 +152,7 @@ Each item is Core's deliberate or native behavior where the official service beh ### Unverified against the official service - The order of errors when one request has several faults, and error, default and payload-limit parity in general. -- A Subagent's function call: Core makes it a required action of the root Turn that runs the child, with `turn_id` set to that Turn. The Turn is `waiting` and the Session `requires_action` while root output may continue, and the `call_id` equals the `call_id` of the child's `function_call` Item. A Subagent's MCP calls are `mcp_call` Items in the child's history. +- A Subagent's function call: Core makes it a required action of the root Turn that runs the child, with `turn_id` set to that Turn. The Turn is `waiting` and the Session `requires_action` while root output may continue, and the `call_id` equals the `call_id` of the child's `function_call` Item. - Whether Subagent child Turns and pending Environment file writes block Session deletion. - The official order between the pending-input error and an unknown result target. - Failure reasons for npm, initial-file and Skill installation have no official sample. diff --git a/contracts/agents-api/message-content.md b/contracts/agents-api/message-content.md index c05da7af9..3ce1632f9 100644 --- a/contracts/agents-api/message-content.md +++ b/contracts/agents-api/message-content.md @@ -58,9 +58,9 @@ An `agent.session.input.tool_result` event carries `success`, an optional nullab Admission (202) does not mean the Harness used the result. The pending call clears when the adapter confirms native application: - **Claude Code** confirms with a live root native tool result that matches the Session, call ID, success flag, exact text, block count and order, with a native image at every image position. Replayed, synthetic and Subagent records do not confirm it. -- **Codex** confirms with the live root dynamic-tool `item/completed` observation that matches thread, Turn, call, function name, status, success flag and the exact ordered content. A write to the Harness alone does not confirm it. +- **Codex** confirms with the live dynamic-tool `item/completed` observation on the thread and Turn that made the call, which must match the call, function name, status, success flag and the exact ordered content. A write to the Harness alone does not confirm it. -Both adapters wait at most 10 seconds for a receipt. A timeout or native release without confirmation leaves application uncertain. Confirmation means the Harness recorded the result, not that the model provider consumed it. Core never replays a result automatically; the submission stays stored for recovery reads. +Both adapters wait at most 10 seconds for a receipt. A timeout or native release without confirmation leaves application uncertain. Confirmation means the Harness recorded the result, not that the model provider consumed it. Core never replays a result automatically; the submission stays stored for recovery reads. A call the native Harness abandons gets no receipt and is withdrawn ([Functions](./execution-tools.md#functions)). ## Runtime boundary diff --git a/contracts/agents-api/subagents.md b/contracts/agents-api/subagents.md index be7eedc3f..db6f4aaf5 100644 --- a/contracts/agents-api/subagents.md +++ b/contracts/agents-api/subagents.md @@ -50,7 +50,7 @@ An adapter freezes root output before child settlement, keeps its native owner a ## Native profiles -A Subagent uses the Agent's function and HTTP MCP tools as the Harness would in a local run, and [Harness capabilities](./harness-capabilities.md#tools) lists each combination. A child's function call is a required action of the root Turn ([Functions](./execution-tools.md#functions)); a child's MCP call is an `mcp_call` Item in its history. Codex children inherit the Agent's HTTP MCP servers, and a forked child inherits its functions while a child with fresh history has none. MiniMax Code projects its children's MCP calls and rejects public functions for the Harness as a whole. Claude Code rejects functions and HTTP MCP with Subagents. +A Subagent uses the Agent's function and HTTP MCP tools as the Harness would in a local run; [Functions](./execution-tools.md#functions) describes how its calls surface, and [Harness capabilities](./harness-capabilities.md#tools) lists each combination. Codex children inherit the Agent's HTTP MCP servers, and a forked child inherits its functions while a child with fresh history has none. **Codex.** The adapter enables the native `multi_agent` feature and maps the concurrency limit to `agents.max_threads`; nesting depth and other native features keep their Codex defaults. It refuses to start if managed requirements force code mode or `multi_agent_v2`, whose collaboration calls leave no persisted receipt. Child Items without a neutral Item type, such as image views and context compaction, are not projected. Close and reopen facts come from direct tool output correlated with the same call's persisted completion, so they need native persisted receipts. Native Turn times have second precision. Child file work can finish under the same owner after the root Turn finishes. Cancellation continues under the same owner after a caller deadline; a later call can confirm settlement without repeating the native interrupt. diff --git a/contracts/agents-api/zh/execution-tools.md b/contracts/agents-api/zh/execution-tools.md index b346fd88f..464bb2d9a 100644 --- a/contracts/agents-api/zh/execution-tools.md +++ b/contracts/agents-api/zh/execution-tools.md @@ -1,7 +1,7 @@ --- title: "执行工具" source: contracts/agents-api/execution-tools.md -source_hash: 09fe0e89ed0d75dac206e08aff4297d3e8f1830a771a6faabc7277070f836546 +source_hash: ac04fcb9091694b3341ea9c2f2fd8af53363731fd6d55cdd14c08ec914658520 --- Agent 在 `tools` 中声明应用函数、控制项和 MCP 服务器，并可在 `text.format` 中声明输出 schema。本契约说明 Core 如何验证声明、哪些内容跨越 Runtime 边界，以及调用方如何恢复待执行操作。[Harness 能力](harness-capabilities.md)列出各 Harness 在不同部署位置支持的操作。原生工作区工具和 Environment Plugin MCP 属于 [Environment](environments.md#skills-plugins-and-environment-mcp)。 @@ -24,11 +24,14 @@ Agent 在 `tools` 中声明应用函数、控制项和 MCP 服务器，并可在 | 相同重试，包括 Turn 结束之后 | 接受 | | 同一调用提交不同结果 | 409 `conflict_error` | | Turn 取消后首次提交结果 | 409 `conflict_error` | +| 为原生 Harness 已放弃的调用提交结果 | 409 `conflict_error` | | 调用方 Session 内未知调用或其他 Turn 的调用 | 400 `invalid_request_error`；待执行操作不变 | | Session 不存在或属于其他范围 | 404 | [Session 输入冲突](sessions-events.md#input-errors)记录准确消息。无效或不支持的内容不能消耗待处理调用。准入与应用结果分离：只有匹配的原生工具结果出现在发起调用的实时 Turn 中，适配器才确认结果（[回执契约](message-content.md#function-results)）。传输写入本身不确认任何事实，确认也不代表提供商已消费结果或外部副作用恰好发生一次。Core 不自动重放结果。 +原生 Harness 可能在应用结果之前放弃调用，例如根 Turn 关闭或中断了发起该调用的 Subagent。该调用随即不再是待执行操作，Turn 继续运行，并在没有待执行操作时回到 `in_progress`；之后为它提交的 `function_call_output` 会像取消后的结果一样被拒绝。被放弃的调用没有应用回执。 + Subagent 发起的函数调用是运行该子级的根 Turn 的待执行操作：`turn_id` 是该根 Turn，`call_id` 等于 [Subagent 路由](subagents.md#public-reads)上该子级 `function_call` Item 的 `call_id`。使用该 `call_id` 和根 Turn ID，以 `function_call_output` / `tool_result` 作答；子级的 Turn ID 不是 Session Turn，会得到“另一个 Turn 的调用”的 400。Subagent 的 HTTP MCP 调用是该子级历史中的 `mcp_call` Item。 ### 待执行操作与恢复 {#required-actions-and-recovery} diff --git a/contracts/agents-api/zh/harness-capabilities.md b/contracts/agents-api/zh/harness-capabilities.md index 26aade8be..bff761590 100644 --- a/contracts/agents-api/zh/harness-capabilities.md +++ b/contracts/agents-api/zh/harness-capabilities.md @@ -47,9 +47,9 @@ source_hash: 6ae7ae05948fe30ce66d86b0a3987c08456392d6a7ed66e71c09ec8b85044843 | [HTTP MCP bearer 凭据](execution-tools.md#http-mcp) | 已验证 | 已验证 | 已验证 | | 必需 MCP 初始化 | 已验证 | 已验证 | 已拒绝 | | [Subagent](subagents.md) | 已验证：托管；已准入：`none`、自托管 | 已验证：托管；已准入：`none`、自托管 | 已验证：托管；已准入：`none`、自托管 | -| Subagent 与函数或 HTTP MCP 同时使用 | 已验证：`none`；已准入：hosted、self-hosted | 已拒绝 | HTTP MCP 已准入；每个 Session 都拒绝函数 | +| Subagent 与函数或 HTTP MCP 同时使用 | 已验证：`none`；已准入：托管、自托管 | 已拒绝 | HTTP MCP 已准入；每个 Session 都拒绝函数 | -Claude 结构化输出要求 medium verbosity，可与 `multi_agent`、Skill、Plugin、能力目录、MCP 和 `tool_search` 组合。Claude 延迟发现可与 MCP、Skill、Plugin、能力目录和结构化输出组合。Subagent 配置不携带函数工具，因此延迟发现与 `multi_agent` 的组合仍被拒绝。`tool_search` 是否对所配置的模型、提供商和 beta 设置生效，由 Claude Code 决定。Claude 在工作区位置需要每种操作的打包桥接功能（`workspace_functions`、`workspace_structured_output`、`workspace_tool_search`、`workspace_mcp_http`）。 +Claude 结构化输出要求 medium verbosity，可与 `multi_agent`、Skill、Plugin、能力目录、MCP 和 `tool_search` 组合。Claude 延迟发现可与 MCP、Skill、Plugin、能力目录和结构化输出组合。Claude Code 不会把函数工具带入 Subagent 配置，因此延迟发现与 `multi_agent` 的组合仍被拒绝。`tool_search` 是否对所配置的模型、提供商和 beta 设置生效，由 Claude Code 决定。Claude 在工作区位置需要每种操作的打包桥接功能（`workspace_functions`、`workspace_structured_output`、`workspace_tool_search`、`workspace_mcp_http`）。 ## Environment 准备 {#environment-preparation} diff --git a/contracts/agents-api/zh/index.md b/contracts/agents-api/zh/index.md index dc22ffe5b..86b7566c0 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: f9e0c35869525870fa49e725a05bfda1d9b4d94af95764728575971ac7cce92e +source_hash: b6c779b6e8264e04e048394ab2a0f1b470639d0e8fe1f79b2fc0bf5564c2bef1 --- Core 实现下方固定版本的 OpenAI Agents API。本台账列出各项资源的覆盖情况、相对于官方服务的所有差异以及已知缺口。 @@ -84,6 +84,7 @@ Core 自身字段位于 `x_agents_core` 中（[Core extensions](../../../docs/zh - 提交给 `none` Session 的输入会同步完成准入；Core 不会模拟官方异步准入窗口。 - 用于恢复等待中 Turn 的函数结果会发出 `turn.in_progress`；取消正在等待函数结果的 Turn 会发出临时 `agent.session.in_progress`。 +- 原生 Harness 放弃的函数调用（例如根关闭或中断子级时被丢弃的 Codex Subagent 调用）会被撤回：它不再是待执行操作，之后为它提交结果会返回 409 `conflict_error`。官方 API 没有撤回机制。 - 在 Turn 中途接入流时，不会发送补发 Item 快照。 - Items 列表会包含进行中和未完成的输出 Items，并保留失败函数结果中已提交的 `output`。 - 错误消息会省略官方消息包含的 call 和 executor ID。 @@ -153,8 +154,8 @@ Core 自身字段位于 `x_agents_core` 中（[Core extensions](../../../docs/zh ### 尚未与官方服务核实的内容 {#unverified-against-the-official-service} -- Subagent 的函数调用：Core 将其作为运行该子级的根 Turn 的待执行操作，`turn_id` 为该 Turn。该 Turn 为 `waiting`、Session 为 `requires_action`，根输出可继续，`call_id` 等于该子级 `function_call` Item 的 `call_id`。Subagent 的 MCP 调用是该子级历史中的 `mcp_call` Item。 - 当一个请求存在多项故障时的错误顺序，以及错误处理、默认值和载荷限制的总体一致性。 +- Subagent 的函数调用：Core 将其作为运行该子级的根 Turn 的待执行操作，`turn_id` 为该 Turn。该 Turn 为 `waiting`、Session 为 `requires_action`，根输出可继续，`call_id` 等于该子级 `function_call` Item 的 `call_id`。 - Subagent 子级 Turns 和待处理的 Environment file 写入是否会阻止 Session 删除。 - 官方服务在待处理输入错误与未知结果目标之间的先后顺序。 - npm、initial-file 和 Skill 安装的失败原因没有官方样本。 diff --git a/contracts/agents-api/zh/message-content.md b/contracts/agents-api/zh/message-content.md index 9876d77ad..27b00e6cf 100644 --- a/contracts/agents-api/zh/message-content.md +++ b/contracts/agents-api/zh/message-content.md @@ -1,7 +1,7 @@ --- title: "消息内容" source: contracts/agents-api/message-content.md -source_hash: e33b8ee25b70fe93691eeb4c84d598fcba0c6703b5e1dfea6c8d5e996e4d098d +source_hash: 71a5c1bb8658f1780631c022a5f28f45b4a55fc4996a428e7372a5920ed693c2 --- 用户消息和函数结果共享同一内容模型：由 `input_text` 与 `input_image` 部分组成的有序列表。Core 按发送形式准确存储消息边界、部分顺序和图像引用，并在用户 Item 中原样返回。不下载、转码或修复媒体。Session 创建的 `input` 与 `events.create` 消息共享验证和准入；[Session、事件与历史](sessions-events.md#send-input)定义准入、请求限制和错误。 @@ -60,9 +60,9 @@ source_hash: e33b8ee25b70fe93691eeb4c84d598fcba0c6703b5e1dfea6c8d5e996e4d098d 准入（202）不表示 Harness 使用了结果。适配器确认原生应用后，待处理调用清除： - **Claude Code** 使用实时根原生工具结果确认，要求匹配 Session、调用 ID、成功标志、准确文本、块数和顺序，且每个图像位置都有原生图像。重放、合成和 Subagent 记录不能确认。 -- **Codex** 使用实时根 dynamic-tool `item/completed` 观察确认，要求匹配线程、Turn、调用、函数名称、状态、成功标志和准确有序内容。仅写入 Harness 不确认。 +- **Codex** 使用发起调用的线程和 Turn 上的实时 dynamic-tool `item/completed` 观察确认，要求匹配调用、函数名称、状态、成功标志和准确有序内容。仅写入 Harness 不确认。 -两适配器等待回执最多 10 秒。超时或原生释放且未确认时，应用保持不确定。确认表示 Harness 记录了结果，不表示模型提供商消费了结果。Core 不自动重放结果；提交仍保留用于恢复读取。 +两适配器等待回执最多 10 秒。超时或原生释放且未确认时，应用保持不确定。确认表示 Harness 记录了结果，不表示模型提供商消费了结果。Core 不自动重放结果；提交仍保留用于恢复读取。原生 Harness 放弃的调用没有回执，会被撤回（[Functions](execution-tools.md#functions)）。 ## Runtime 边界 {#runtime-boundary} diff --git a/contracts/agents-api/zh/subagents.md b/contracts/agents-api/zh/subagents.md index 03fee3bfe..d4aefa46d 100644 --- a/contracts/agents-api/zh/subagents.md +++ b/contracts/agents-api/zh/subagents.md @@ -1,7 +1,7 @@ --- title: "子 Agent" source: contracts/agents-api/subagents.md -source_hash: c70156da39cb9eac2a6e86e7b6807ef4fef0315c59c84cc84f506f8dcf905ec6 +source_hash: 735f0570a1c9d5feaebd6b38aacb186c9982aa739f7a835ddb2e02a071fc2196 --- 启用 `multi_agent.enabled` 后，Harness 可以启动原生子 Agent。Core 通过锁定版本的 SDK（见 [`upstream.json`](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/contracts/agents-api/upstream.json)）中的六项 Subagent 读取操作公开它们，并根据适配器观察结果记录它们。当 `multi_agent.enabled=false` 时，Runtime 会移除原生子 Agent 工具。[Harness capabilities](harness-capabilities.md) 列出各 Harness 在哪些组合中支持子 Agent。 @@ -52,7 +52,7 @@ Core 根据已授权的 Session 绑定关系分配合公开 ID 和所有权。 ## 原生配置 {#native-profiles} -Subagent 像本地运行该 Harness 一样使用 Agent 的函数和 HTTP MCP 工具，各组合见 [Harness capabilities](harness-capabilities.md#tools)。子级的函数调用是根 Turn 的待执行操作（[Functions](execution-tools.md#functions)）；子级的 MCP 调用是其历史中的 `mcp_call` Item。Codex 子级继承 Agent 的 HTTP MCP 服务器；fork 出的子级继承其函数，使用全新历史的子级没有函数。MiniMax Code 会投影其子级的 MCP 调用，并对整个 Harness 拒绝公开函数。Claude Code 在使用 Subagent 时拒绝函数和 HTTP MCP。 +Subagent 像本地运行该 Harness 一样使用 Agent 的函数和 HTTP MCP 工具；其调用如何呈现见 [Functions](execution-tools.md#functions)，各组合见 [Harness capabilities](harness-capabilities.md#tools)。Codex 子级继承 Agent 的 HTTP MCP 服务器；fork 出的子级继承其函数，使用全新历史的子级没有函数。 **Codex.** 适配器会启用原生 `multi_agent` 特性，并把并发限制映射到 `agents.max_threads`；嵌套深度和其他原生特性保持 Codex 默认值。如果托管配置要求强制启用代码模式或 `multi_agent_v2`（它们的协作调用不留下持久化回执），则拒绝启动。没有中立 Item 类型的子级 Item（例如图像查看和上下文压缩）不会被投影。关闭和重新打开的事实来自直接工具输出，并与同一次调用的持久化完成记录相关联，因此需要原生持久化回执。原生 Turn 时间具有秒级精度。根级 Turn 完成后，子级文件工作可以在同一所有者下完成。超过调用方截止时间后，取消操作仍会在同一所有者下继续；后续调用可以确认已经收敛，而无需重复执行原生中断。 diff --git a/docs/runtime-protocol.md b/docs/runtime-protocol.md index e739e3d56..1eef60f76 100644 --- a/docs/runtime-protocol.md +++ b/docs/runtime-protocol.md @@ -110,7 +110,7 @@ The linked source files define the required fields, validators, limits and finit | `execution_prepare`, `execution_start`, `execution_release` | `preparation_status` | [Execution admission](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/preparation.go) | | `prompt_cancel` | `delta`, `thinking`, `output_message`, `tool_call`, `usage`, `error`, `done`, `heartbeat`, `interaction_decision_ack` | [Requests](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/outbound.go), [events and capabilities](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/inbound.go) | | `prompt_steer` | `prompt_steer_ack` | [Active input receipts](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/steering.go) | -| `function_result` | `function_call`, `interaction_decision_ack` | [Function calls](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/functions.go) | +| `function_result` | `function_call`, `function_call_withdrawn`, `interaction_decision_ack` | [Function calls](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/functions.go) | | `workspace_read`, `workspace_write`, `workspace_export` | Matching `*_result` | [Read](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/workspace_read.go), [write](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/workspace_write.go), [export](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/workspace_export.go) | | `environment_quiesce`, `environment_resume` | `environment_quiesced`, `environment_resumed` | [Suspension fencing](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/suspend.go) | @@ -118,6 +118,8 @@ Initial, prepared and active input use the same [ordered MessageInput](https://g A `function_call` carries `native_id` and `turn_id` when a Subagent made the call, with the meaning they have in `subagent_turn`; both are empty for a call of the Run itself. The adapter sends the child's `subagent_identity` and `subagent_turn` before the call, and Core answers with `function_result` under the same `call_id`. +A `function_call_withdrawn` carries the `call_id`, `native_id` and `turn_id` of a `function_call` and tells that the native Harness will never apply a result for it, for example because a Codex Subagent Turn ended and native dropped its request. Core stops requiring an answer for the call and rejects a later result for it; a `function_result` already in flight is acknowledged as not applied, and the adapter answers it with `not_pending`. + Usage frames and the final usage snapshot each carry the cumulative measurement of the current execution and replace the previous snapshot; never add them. An absent measurement is unknown, not zero. ## Preparation and execution order diff --git a/docs/zh/runtime-protocol.md b/docs/zh/runtime-protocol.md index bc60a3897..1a2e7c697 100644 --- a/docs/zh/runtime-protocol.md +++ b/docs/zh/runtime-protocol.md @@ -1,7 +1,7 @@ --- title: "Core–Runtime 协议" source: docs/runtime-protocol.md -source_hash: dfd5005d5d5a51df20d1407fe897237302c7e16f0ad6e6decd0910d181e3f039 +source_hash: 7dee36cd45d967980ed9a6792b6facff9916791046968f7733376033ca963efb --- Core–Runtime 协议连接 Core 与 Runtime daemon，定义 daemon 连接上消息的含义和顺序。托管和自托管 Runtime 使用同一协议。 @@ -112,7 +112,7 @@ Core 对 `usage` 和 `resume` 没有准入规则。 | `execution_prepare`, `execution_start`, `execution_release` | `preparation_status` | [执行准入](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/preparation.go) | | `prompt_cancel` | `delta`, `thinking`, `output_message`, `tool_call`, `usage`, `error`, `done`, `heartbeat`, `interaction_decision_ack` | [请求](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/outbound.go)、[事件与能力](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/inbound.go) | | `prompt_steer` | `prompt_steer_ack` | [活动输入回执](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/steering.go) | -| `function_result` | `function_call`, `interaction_decision_ack` | [Function 调用](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/functions.go) | +| `function_result` | `function_call`, `function_call_withdrawn`, `interaction_decision_ack` | [Function 调用](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/functions.go) | | `workspace_read`, `workspace_write`, `workspace_export` | 对应的 `*_result` | [读取](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/workspace_read.go)、[写入](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/workspace_write.go)、[导出](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/workspace_export.go) | | `environment_quiesce`, `environment_resume` | `environment_quiesced`, `environment_resumed` | [暂停 fencing](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/internal/agentdaemon/proto/suspend.go) | @@ -120,6 +120,8 @@ Core 对 `usage` 和 `resume` 没有准入规则。 Subagent 发起调用时，`function_call` 携带 `native_id` 和 `turn_id`，含义与 `subagent_turn` 中相同；Run 自身发起的调用两者均为空。Adapter 在调用之前发送该子级的 `subagent_identity` 和 `subagent_turn`，Core 用同一个 `call_id` 的 `function_result` 作答。 +`function_call_withdrawn` 携带某个 `function_call` 的 `call_id`、`native_id` 和 `turn_id`，表示原生 Harness 永远不会为它应用结果，例如 Codex Subagent 的 Turn 已结束且原生丢弃了其请求。Core 不再要求回答该调用，并拒绝之后为它提交的结果；已在传输中的 `function_result` 会以未应用确认，adapter 以 `not_pending` 作答。 + Usage frame 和最终 usage snapshot 都携带当前执行的累计测量，替换之前的快照；不要相加。缺失的测量表示未知，不是零。 ## 准备与执行顺序 {#preparation-and-execution-order} From 06fca2d82482835210f5c899e7be1779ff507efd Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 19 Oct 2026 16:17:53 +0000 Subject: [PATCH 4/7] Leave an abandoned in-flight result to its withdrawal --- docs/runtime-protocol.md | 2 +- docs/zh/runtime-protocol.md | 4 ++-- services/core/internal/execution/delivery.go | 2 +- services/core/internal/execution/functions.go | 19 +++++++++---------- 4 files changed, 13 insertions(+), 14 deletions(-) diff --git a/docs/runtime-protocol.md b/docs/runtime-protocol.md index 1eef60f76..b906a10e2 100644 --- a/docs/runtime-protocol.md +++ b/docs/runtime-protocol.md @@ -118,7 +118,7 @@ Initial, prepared and active input use the same [ordered MessageInput](https://g A `function_call` carries `native_id` and `turn_id` when a Subagent made the call, with the meaning they have in `subagent_turn`; both are empty for a call of the Run itself. The adapter sends the child's `subagent_identity` and `subagent_turn` before the call, and Core answers with `function_result` under the same `call_id`. -A `function_call_withdrawn` carries the `call_id`, `native_id` and `turn_id` of a `function_call` and tells that the native Harness will never apply a result for it, for example because a Codex Subagent Turn ended and native dropped its request. Core stops requiring an answer for the call and rejects a later result for it; a `function_result` already in flight is acknowledged as not applied, and the adapter answers it with `not_pending`. +A `function_call_withdrawn` carries the `call_id`, `native_id` and `turn_id` of a `function_call` and tells that the native Harness will never apply a result for it, for example because a Codex Subagent Turn ended and native dropped its request. Core stops requiring an answer for the call and rejects a later result for it. The adapter acknowledges a `function_result` already in flight with `not_pending`, and Core then waits for the withdrawal or the Turn's end instead of failing the Turn or resending the result. Usage frames and the final usage snapshot each carry the cumulative measurement of the current execution and replace the previous snapshot; never add them. An absent measurement is unknown, not zero. diff --git a/docs/zh/runtime-protocol.md b/docs/zh/runtime-protocol.md index 1a2e7c697..3d1f205fa 100644 --- a/docs/zh/runtime-protocol.md +++ b/docs/zh/runtime-protocol.md @@ -1,7 +1,7 @@ --- title: "Core–Runtime 协议" source: docs/runtime-protocol.md -source_hash: 7dee36cd45d967980ed9a6792b6facff9916791046968f7733376033ca963efb +source_hash: 8c334fe24521cf20a300c447b6b81d4aa0f7bc786bd89bc45722187d0c004acf --- Core–Runtime 协议连接 Core 与 Runtime daemon，定义 daemon 连接上消息的含义和顺序。托管和自托管 Runtime 使用同一协议。 @@ -120,7 +120,7 @@ Core 对 `usage` 和 `resume` 没有准入规则。 Subagent 发起调用时，`function_call` 携带 `native_id` 和 `turn_id`，含义与 `subagent_turn` 中相同；Run 自身发起的调用两者均为空。Adapter 在调用之前发送该子级的 `subagent_identity` 和 `subagent_turn`，Core 用同一个 `call_id` 的 `function_result` 作答。 -`function_call_withdrawn` 携带某个 `function_call` 的 `call_id`、`native_id` 和 `turn_id`，表示原生 Harness 永远不会为它应用结果，例如 Codex Subagent 的 Turn 已结束且原生丢弃了其请求。Core 不再要求回答该调用，并拒绝之后为它提交的结果；已在传输中的 `function_result` 会以未应用确认，adapter 以 `not_pending` 作答。 +`function_call_withdrawn` 携带某个 `function_call` 的 `call_id`、`native_id` 和 `turn_id`，表示原生 Harness 永远不会为它应用结果，例如 Codex Subagent 的 Turn 已结束且原生丢弃了其请求。Core 不再要求回答该调用，并拒绝之后为它提交的结果。adapter 以 `not_pending` 确认已在传输中的 `function_result`，Core 随后等待撤回或 Turn 结束，既不让 Turn 失败，也不重发该结果。 Usage frame 和最终 usage snapshot 都携带当前执行的累计测量，替换之前的快照；不要相加。缺失的测量表示未知，不是零。 diff --git a/services/core/internal/execution/delivery.go b/services/core/internal/execution/delivery.go index 1ec0e631b..9a815c11a 100644 --- a/services/core/internal/execution/delivery.go +++ b/services/core/internal/execution/delivery.go @@ -99,7 +99,7 @@ func (d *Dispatcher) deliver(ctx context.Context, tenantID, sessionID string, pe var pending *pendingInput var cancelSent time.Time var cancelReply <-chan cancellationResult - functions := &functionExchange{kind: request.AgentKind, turns: d.SessionsReader, sessions: d.sessionExecution, tenant: tenantID, session: sessionID, turn: runID, tools: request.FunctionTools} + functions := &functionExchange{kind: request.AgentKind, turns: d.SessionsReader, sessions: d.sessionExecution, tenant: tenantID, session: sessionID, turn: runID, tools: request.FunctionTools, abandoned: map[string]bool{}} done := false cancelCtx, stopCancellation := context.WithCancel(ctx) defer stopCancellation() diff --git a/services/core/internal/execution/functions.go b/services/core/internal/execution/functions.go index 96f7d4b2d..19ef8d43d 100644 --- a/services/core/internal/execution/functions.go +++ b/services/core/internal/execution/functions.go @@ -6,7 +6,6 @@ import ( "encoding/json" "errors" "fmt" - "slices" "strings" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" @@ -54,6 +53,9 @@ type functionExchange struct { tools []proto.FunctionTool callID string reply <-chan functionReply + // abandoned holds the calls whose result the native Harness no longer + // waited for. Their withdrawal, or the Turn's end, settles them. + abandoned map[string]bool } func (f *functionExchange) record(ctx context.Context, env proto.Envelope) error { @@ -96,7 +98,7 @@ func (f *functionExchange) start(ctx context.Context, peer *runtimegateway.Sessi return err } for _, call := range calls { - if len(call.Result) == 0 { + if len(call.Result) == 0 || f.abandoned[call.CallID] { continue } result, err := functionResult(call) @@ -128,15 +130,12 @@ func (f *functionExchange) confirm(ctx context.Context, reply functionReply) err return reply.err } if !reply.ack.Applied { - // A result in flight when the native Harness abandoned the call is moot. - calls, err := f.sessions.PendingFunctionCalls(ctx, f.tenant, f.session, f.turn) - if err != nil { - return err + if reply.ack.ErrorCode != "not_pending" { + return fmt.Errorf("function result not applied: %s", reply.ack.ErrorCode) } - if !slices.ContainsFunc(calls, func(call sessions.FunctionCall) bool { return call.CallID == id }) { - return nil - } - return fmt.Errorf("function result not applied: %s", reply.ack.ErrorCode) + // The withdrawal can reach this loop after the acknowledgement. + f.abandoned[id] = true + return nil } return f.unlessCancelling(ctx, f.sessions.ConfirmFunctionResult(ctx, f.tenant, f.session, f.turn, id)) } From 5eeef64259cddfc3e407107598142fb8262458ee Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 19 Oct 2026 18:05:50 +0000 Subject: [PATCH 5/7] Restore the pending index on downgrade and expect withdrawal after not_pending --- services/core/migrations/000103_function_call_withdrawn.sql | 4 +++- services/core/tests/integration/function_execution_test.go | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/services/core/migrations/000103_function_call_withdrawn.sql b/services/core/migrations/000103_function_call_withdrawn.sql index 861c7028b..2c7f2863e 100644 --- a/services/core/migrations/000103_function_call_withdrawn.sql +++ b/services/core/migrations/000103_function_call_withdrawn.sql @@ -6,4 +6,6 @@ DROP INDEX function_calls_pending_idx; CREATE INDEX function_calls_pending_idx ON function_calls(session_id, turn_id) WHERE NOT applied AND NOT withdrawn; -- +goose Down -SELECT 1; +DROP INDEX function_calls_pending_idx; +CREATE INDEX function_calls_pending_idx ON function_calls(session_id, turn_id) WHERE NOT applied; +ALTER TABLE function_calls DROP COLUMN withdrawn; diff --git a/services/core/tests/integration/function_execution_test.go b/services/core/tests/integration/function_execution_test.go index c9aff3196..f44935819 100644 --- a/services/core/tests/integration/function_execution_test.go +++ b/services/core/tests/integration/function_execution_test.go @@ -202,7 +202,7 @@ func TestExecutionFunctionsCancellationAndUnconfirmedResults(t *testing.T) { if reply.Success || len(reply.Content) != 1 || *reply.Content[0].Text != "tool failed" { t.Fatal(reply) } - status := sessions.TurnFailed + status := sessions.TurnCompleted if cancel { if _, err := requestCancel(t.Context(), h.s, h.tenant, h.session.ID, "cancel"); err != nil { t.Fatal(err) @@ -219,6 +219,8 @@ func TestExecutionFunctionsCancellationAndUnconfirmedResults(t *testing.T) { status = sessions.TurnCancelled } else { h.write(input.TurnID, proto.TypeInteractionDecisionAck, proto.InteractionDecisionAckPayload{DeliveryID: reply.DeliveryID, ErrorCode: "not_pending"}) + h.write(input.TurnID, proto.TypeFunctionCallWithdrawn, proto.FunctionCallWithdrawnPayload{CallID: "a"}) + h.write(input.TurnID, proto.TypeDone, proto.DonePayload{Content: "done"}) } h.finished(result, status) if cancel { From 6002f9f1cd8eda366184a52d1aa1e2d2bfafbc54 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 19 Oct 2026 18:19:20 +0000 Subject: [PATCH 6/7] Accept an identical retry after withdrawal and flush before it --- apps/daemon/internal/agent/codex/functions.go | 2 +- contracts/agents-api/execution-tools.md | 4 ++-- contracts/agents-api/zh/execution-tools.md | 6 +++--- services/core/internal/execution/delivery.go | 11 +++++------ services/core/internal/sessions/functions.go | 12 +++++------- .../core/internal/sessions/functions_test.go | 16 +++++++++------- 6 files changed, 25 insertions(+), 26 deletions(-) diff --git a/apps/daemon/internal/agent/codex/functions.go b/apps/daemon/internal/agent/codex/functions.go index c678d994f..925f50e3a 100644 --- a/apps/daemon/internal/agent/codex/functions.go +++ b/apps/daemon/internal/agent/codex/functions.go @@ -121,7 +121,7 @@ type functionContent struct { func (s *Session) stopFunctionCalls(rootOnly bool) { if s.functions != nil { s.functions.mu.Lock() - s.functions.closed = !rootOnly + s.functions.closed = s.functions.closed || !rootOnly for id, pending := range s.functions.pending { if !rootOnly || pending.thread == s.currentThreadID() { pending.receipt <- agent.ErrUnknownFunctionCall diff --git a/contracts/agents-api/execution-tools.md b/contracts/agents-api/execution-tools.md index b82f1e655..6a7a79c3b 100644 --- a/contracts/agents-api/execution-tools.md +++ b/contracts/agents-api/execution-tools.md @@ -19,10 +19,10 @@ A function declaration requires `name`, `description` and a JSON Schema in `para | Case | Response | | --- | --- | -| Identical retry, also after the Turn ends | Accepted | +| Identical retry, also after the Turn ends or the call is abandoned | Accepted | | Different result for the same call | 409 `conflict_error` | | First result after the Turn was cancelled | 409 `conflict_error` | -| Result for a call the native Harness abandoned | 409 `conflict_error` | +| First result for a call the native Harness abandoned | 409 `conflict_error` | | Unknown call, or a call of another Turn, in the caller's Session | 400 `invalid_request_error`; the pending action is unchanged | | Missing or foreign Session | 404 | diff --git a/contracts/agents-api/zh/execution-tools.md b/contracts/agents-api/zh/execution-tools.md index 464bb2d9a..1b653d397 100644 --- a/contracts/agents-api/zh/execution-tools.md +++ b/contracts/agents-api/zh/execution-tools.md @@ -1,7 +1,7 @@ --- title: "执行工具" source: contracts/agents-api/execution-tools.md -source_hash: ac04fcb9091694b3341ea9c2f2fd8af53363731fd6d55cdd14c08ec914658520 +source_hash: 5e14ecb8add3ada10039f01ccdb0125c8f545c67f759aad74e973b7ac80cf00e --- Agent 在 `tools` 中声明应用函数、控制项和 MCP 服务器，并可在 `text.format` 中声明输出 schema。本契约说明 Core 如何验证声明、哪些内容跨越 Runtime 边界，以及调用方如何恢复待执行操作。[Harness 能力](harness-capabilities.md)列出各 Harness 在不同部署位置支持的操作。原生工作区工具和 Environment Plugin MCP 属于 [Environment](environments.md#skills-plugins-and-environment-mcp)。 @@ -21,10 +21,10 @@ Agent 在 `tools` 中声明应用函数、控制项和 MCP 服务器，并可在 | 情况 | 响应 | | --- | --- | -| 相同重试，包括 Turn 结束之后 | 接受 | +| 相同重试，包括 Turn 结束或调用被放弃之后 | 接受 | | 同一调用提交不同结果 | 409 `conflict_error` | | Turn 取消后首次提交结果 | 409 `conflict_error` | -| 为原生 Harness 已放弃的调用提交结果 | 409 `conflict_error` | +| 为原生 Harness 已放弃的调用首次提交结果 | 409 `conflict_error` | | 调用方 Session 内未知调用或其他 Turn 的调用 | 400 `invalid_request_error`；待执行操作不变 | | Session 不存在或属于其他范围 | 404 | diff --git a/services/core/internal/execution/delivery.go b/services/core/internal/execution/delivery.go index 9a815c11a..017c94dfc 100644 --- a/services/core/internal/execution/delivery.go +++ b/services/core/internal/execution/delivery.go @@ -217,17 +217,16 @@ func (d *Dispatcher) deliver(ctx context.Context, tenantID, sessionID string, pe case proto.TypeDone: // Done closes the event subscription; application receipts have a separate waiter. done, upstream = true, nil - case proto.TypeFunctionCall: + case proto.TypeFunctionCall, proto.TypeFunctionCallWithdrawn: if err := journal.flush(ctx); err != nil { result.ErrorCode = "event_persistence_failed" return } - if err := functions.record(ctx, env); err != nil { - result.ErrorCode = "function_call_invalid" - return + settle := functions.record + if env.Type == proto.TypeFunctionCallWithdrawn { + settle = functions.withdraw } - case proto.TypeFunctionCallWithdrawn: - if err := functions.withdraw(ctx, env); err != nil { + if err := settle(ctx, env); err != nil { result.ErrorCode = "function_call_invalid" return } diff --git a/services/core/internal/sessions/functions.go b/services/core/internal/sessions/functions.go index 7d8008bdd..7a666664b 100644 --- a/services/core/internal/sessions/functions.go +++ b/services/core/internal/sessions/functions.go @@ -65,20 +65,18 @@ type FunctionResultMatch struct { } // decideFunctionResult decides whether to store a submitted result for a call -// the Turn has. A withdrawn call takes no result. An identical retry of a saved result stores nothing, even after -// the Turn ended; a different result conflicts with the saved one. A new result -// needs a Turn that accepts results. +// the Turn has. An identical retry of a saved result stores nothing, even after +// the Turn ended or the call was withdrawn; a different result conflicts with +// the saved one. A new result needs a call that was not withdrawn and a Turn +// that accepts results. func decideFunctionResult(turn Turn, match FunctionResultMatch) (bool, error) { - if match.Withdrawn { - return false, ErrTurnConflict - } if match.Submitted { if !match.Matches { return false, ErrFunctionResultConflict } return false, nil } - if !acceptsFunctionResults(turn) { + if match.Withdrawn || !acceptsFunctionResults(turn) { return false, ErrTurnConflict } return true, nil diff --git a/services/core/internal/sessions/functions_test.go b/services/core/internal/sessions/functions_test.go index 71600da65..c220b5c48 100644 --- a/services/core/internal/sessions/functions_test.go +++ b/services/core/internal/sessions/functions_test.go @@ -170,13 +170,15 @@ func TestDecideFunctionResult(t *testing.T) { store bool err error }{ - "new result": {turnWith(TurnWaiting), FunctionResultMatch{Recorded: true}, true, nil}, - "identical retry": {turnWith(TurnWaiting), same, false, nil}, - "identical retry after end": {turnWith(TurnCompleted), same, false, nil}, - "different result": {turnWith(TurnWaiting), saved, false, ErrFunctionResultConflict}, - "different result after end": {turnWith(TurnCancelled), saved, false, ErrFunctionResultConflict}, - "new result while cancelling": {cancelling(TurnWaiting), FunctionResultMatch{Recorded: true}, false, ErrTurnConflict}, - "new result after the Turn end": {turnWith(TurnFailed), FunctionResultMatch{Recorded: true}, false, ErrTurnConflict}, + "new result": {turnWith(TurnWaiting), FunctionResultMatch{Recorded: true}, true, nil}, + "identical retry": {turnWith(TurnWaiting), same, false, nil}, + "identical retry after end": {turnWith(TurnCompleted), same, false, nil}, + "different result": {turnWith(TurnWaiting), saved, false, ErrFunctionResultConflict}, + "different result after end": {turnWith(TurnCancelled), saved, false, ErrFunctionResultConflict}, + "new result while cancelling": {cancelling(TurnWaiting), FunctionResultMatch{Recorded: true}, false, ErrTurnConflict}, + "new result after the Turn end": {turnWith(TurnFailed), FunctionResultMatch{Recorded: true}, false, ErrTurnConflict}, + "identical retry after withdrawal": {turnWith(TurnInProgress), FunctionResultMatch{Recorded: true, Submitted: true, Matches: true, Withdrawn: true}, false, nil}, + "new result after withdrawal": {turnWith(TurnInProgress), FunctionResultMatch{Recorded: true, Withdrawn: true}, false, ErrTurnConflict}, } { store, err := decideFunctionResult(test.turn, test.match) if store != test.store || !errors.Is(err, test.err) || (test.err == nil && err != nil) { From 1a1b7062cc9db1c8766cd6b57f6f0423704c70b7 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 19 Oct 2026 18:25:30 +0000 Subject: [PATCH 7/7] Bump the Core-Runtime protocol to 0.20.0 --- internal/agentdaemon/proto/version.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/agentdaemon/proto/version.go b/internal/agentdaemon/proto/version.go index 072d9e099..7280dfe4c 100644 --- a/internal/agentdaemon/proto/version.go +++ b/internal/agentdaemon/proto/version.go @@ -2,7 +2,7 @@ package proto // Version identifies the complete Core–Runtime wire contract. Change it when // removing or changing a payload or its semantics; deploy both endpoints together. -const Version = "0.19.0" +const Version = "0.20.0" // VersionCompatible accepts only this contract. Patch drift, prerelease suffixes // and malformed versions do not select an implicit compatibility path.