From 0cc604fb6a7cbc4f2276ffdb28017c5a1abbecca Mon Sep 17 00:00:00 2001 From: Sertac Ozercan Date: Mon, 14 Sep 2026 12:42:58 -0700 Subject: [PATCH] feat: support brokered tool approvals Keep MCP tool calls open for the Orka approval window while preserving model and discovery deadlines. Forward safe final approval outcomes to AgentKit and verify cancellation, counted execution, and fenced continuation. Signed-off-by: Sertac Ozercan --- docs/harness-v2.md | 106 ++++++- internal/acp/approval_test.go | 278 +++++++++++++++++++ internal/acp/config.go | 4 + internal/acp/mcp.go | 17 +- internal/broker/agentkit.go | 37 ++- internal/broker/agentkit_integration_test.go | 70 ++++- internal/broker/agentkit_test.go | 37 ++- internal/broker/approval_test.go | 102 +++++++ 8 files changed, 626 insertions(+), 25 deletions(-) create mode 100644 internal/acp/approval_test.go create mode 100644 internal/broker/approval_test.go diff --git a/docs/harness-v2.md b/docs/harness-v2.md index 52db512..7b5522c 100644 --- a/docs/harness-v2.md +++ b/docs/harness-v2.md @@ -47,8 +47,9 @@ controller epoch, workspace, and MCP settings. The child accepts text prompts and resource links represented as text. It uses the supervisor's HTTP MCP server for tools. It does not support image, audio, embedded-resource, permission, terminal, filesystem, or session-load -ACP requests. Keep approval-required tools empty. The Task must use the -runtime profile's exact brokered tool allowlist. +ACP requests. Orka can require human approval for brokered tools as described +below. This does not enable native ACP permission requests. The Task must use +the runtime profile's exact brokered tool allowlist. The hosted agent's static function names must exactly match those MCP tools. Use Orka's built-in names or the names of its Kubernetes Tool resources. The @@ -137,10 +138,11 @@ AgentKit expects a result envelope with `approved` and either `output` or An explicit `isError: false` becomes `approved: true` with the text and structured content preserved under `output`. `isError: true` becomes `approved: false` with a `brokered_tool_error` code and the validated error -text. Malformed results and missing error flags are rejected. MCP authorization -failures abort the tool call before any result is sent to AgentKit. The -`approved` field is AgentKit's result format; it does not report human approval. -Approval-required tools remain unsupported by this adapter. +text. Orka's recognized structured approval and execution outcomes retain their +specific code and use a fixed safe message. Malformed results and missing error +flags are rejected. MCP authorization failures abort the tool call before any +result is sent to AgentKit. The `approved` field is AgentKit's result format; +it does not report human approval, and `false` is a final outcome. This is the existing AgentKit shared-secret contract. It authenticates the broker's continuation route; it is not a signed execution receipt. The broker's @@ -154,6 +156,79 @@ Leave the variable unset for other Hosted Agents. Their function output format and requests remain unchanged. AgentKit must also support repeated tool rounds for workflows that need several lookups before answering. +## Human approvals + +Use matched builds containing [Orka's brokered approval support](https://github.com/orka-agents/orka/issues/582), +[this bridge's approval integration](https://github.com/orka-agents/agent-runtime-foundry/issues/4), +and [AgentKit's approval compatibility](https://github.com/sozercan/agentkit/issues/26). +Older bridges impose a two-minute tool limit. The Orka v2 runtime must advertise +`supportsBrokeredToolApprovals: true` for the qualified provider combination; +native `supportsPermissions` remains false. Validate the exact pinned Foundry +agent version and gateway before enabling approval-required tools. + +Orka owns the review, authorized reviewers, saved decision, execution, and audit. +Configure an automatic lookup and a harmless counted action requiring approval +in Orka's tool policy. When the agent proposes the action, Orka's Task approval +API and panel show the proposed operation and safe input preview. The action's +execution count remains zero until an authorized reviewer approves it. Each later +approval-required action needs its own decision, including another invocation +of the same tool. + +The existing MCP `tools/call` request stays open while the review is pending. +There is no interim result, polling model request, or resubmission of the prompt. +An approved call returns its actual result to the original pending function call +and previous response. Separate runtime sessions can progress during the wait. +Each child still admits at most two concurrent MCP calls. + +| Operation | Maximum duration | +| --- | --- | +| Human review in Orka | 600 seconds | +| Approved tool execution in Orka | 240 seconds | +| One MCP `tools/call`, including review, execution, and delivery | 900 seconds | +| Model requests and MCP discovery | 120 seconds | +| Local connection and TLS handshake | 10 seconds each | +| Broker provisioning or cleanup operation | 45 seconds | + +These are upper bounds. Task and session deadlines, cancellation, and loss of +the live broker lease can end a wait sooner. The supervisor must continue +renewing its lease while a tool waits; each lease grant remains limited to five +minutes. A longer tool wait does not lengthen any model, discovery, or cleanup +request. The bridge's 900-second limit is fixed in the qualified adapter build; +there is no timeout environment variable to pass to the child. + +Configure both the pinned Foundry version's session idle timeout and AgentKit's +pending-response TTL to at least 1,800 seconds for the full review window. +Set `AGENTKIT_FOUNDRY_RESPONSE_STATE_TTL_SECONDS=1800` on the hosted AgentKit +process. It exposes the configured TTL as `foundryResponses.stateTtlSeconds` on +`/readiness`; use `AGENTKIT_FOUNDRY_RESPONSE_STATE_FILE` for its supported private, +single-writer persistent storage. [Foundry sessions](https://learn.microsoft.com/azure/foundry/agents/how-to/manage-hosted-sessions) +default to a 900-second idle timeout and accept 120 through 3,600 seconds. +Changing that setting requires a new agent version. Neither the default idle +timeout nor a 900-second AgentKit TTL leaves enough margin for the maximum tool +wait and model continuation. A suspended or lost hosted process must not cause +the bridge to repeat an action whose outcome is unknown. + +Orka returns final errors with `isError: true` and an allowlisted +`structuredContent.code`, also marked `isError: true`. The AgentKit continuation +uses `approved: false` with that code and a fixed message: + +| Code | Meaning | +| --- | --- | +| `approval_declined` | The reviewer declined the action; it did not execute. | +| `approval_expired` | The review expired before execution. | +| `approval_cancelled` | The unstarted action was cancelled. | +| `approval_stale` | The recorded decision no longer authorizes the proposed action. | +| `tool_execution_failed` | An approved execution failed. | +| `tool_outcome_unknown` | Execution may have happened; do not repeat the action. | + +The bridge never interprets tool text as an approval decision. Recognized codes +discard raw error text and private metadata when converting to AgentKit's +envelope. Other admitted tool errors keep the existing `brokered_tool_error` +format. HTTP or JSON-RPC authorization errors remain fatal protocol errors. +Cancellation and transport failures close the wait without a continuation; +a late decision cannot reopen that ACP session. Broker restart closes the old +prompt's authority while retaining its ownership evidence. + ## Lifecycle guarantees and limits The broker persists a random caller-chosen remote session ID before sending @@ -221,7 +296,9 @@ go test ./internal/broker -run TestBrokerAgentKitHosted -count=1 -v This runs the production hosted AgentKit server and model loop, the Foundry ACP entrypoint over pipes, and the lifecycle broker. It checks two sequential tools, tool-error recovery, authorization denial, response identity changes, and proof -isolation. A gateway that strips the proof is also tested to verify that no model +isolation. Counted approval fixtures hold the first call, verify zero execution +and no model continuation, then exercise approval, decline, expiry, and a tool +failure after approval. A gateway that strips the proof is also tested to verify that no model resume occurs. Cancellation cases hold the model connection open, wait for an early hosted response acknowledgement, and verify that disconnect and lease expiry close the model connection and allow proven retirement. A gateway that @@ -244,3 +321,18 @@ The model, MCP backend, supervisor context stamping, and Azure session-management API are local fixtures. It requires no Azure or model credentials and does not validate a deployed Orka controller or the public Foundry gateway. These tests are skipped when `AGENTKIT_SOURCE_DIR` is unset. + +The ordinary Go suite also exercises the full 900-second transport budget with +virtual time, approval cancellation, independent sessions, distinct decisions +for sequential calls, lost tool results, rolling leases, and broker restart. +These tests use fixtures for the human decision and do not establish that the +configured Foundry gateway supports a live review. + +For deployment acceptance, run Orka's real Task and approval APIs against the +configured Foundry gateway with a disposable Task and a harmless counted tool. +Record the visible pending approval with count zero, the saved decision, count +one after approval, continuation of the original conversation, and cleanup. +Repeat with a declined review and a cancelled Task; both must keep the unstarted +action's count at zero, including after a late decision. Keep the record free of +credentials and private review metadata. Passing the local fixtures does not +complete this live check. diff --git a/internal/acp/approval_test.go b/internal/acp/approval_test.go new file mode 100644 index 0000000..886fae4 --- /dev/null +++ b/internal/acp/approval_test.go @@ -0,0 +1,278 @@ +package acp + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "io" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/orka-agents/agent-runtime-foundry/internal/foundry" +) + +// Real HTTP encoding and the production transport run over net.Pipe, so virtual +// time can exercise the actual two- and fifteen-minute client deadlines. +func TestACPHeldToolCallTimeouts(t *testing.T) { + for _, tc := range []struct { + name string + method string + delay time.Duration + cancelAt time.Duration + wantTime time.Duration + wantResult bool + }{ + {"review_over_two_minutes", "tools/call", 130 * time.Second, 0, 130 * time.Second, true}, + {"review_and_execution", "tools/call", 840 * time.Second, 0, 840 * time.Second, true}, + {"tool_hard_limit", "tools/call", 901 * time.Second, 0, 900 * time.Second, false}, + {"task_cancelled", "tools/call", 840 * time.Second, 130 * time.Second, 130 * time.Second, false}, + {"discovery_keeps_limit", "tools/list", 121 * time.Second, 0, 120 * time.Second, false}, + {"model_keeps_limit", "model", 121 * time.Second, 0, 120 * time.Second, false}, + } { + t.Run(tc.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if tc.cancelAt != 0 { + go func() { time.Sleep(tc.cancelAt); cancel() }() + } + client := newACPHTTPClient() + defer client.CloseIdleConnections() + var wg sync.WaitGroup + var calls atomic.Int32 + client.Transport.(*http.Transport).DialContext = func(context.Context, string, string) (net.Conn, error) { + left, right := net.Pipe() + wg.Go(func() { + defer right.Close() //nolint:errcheck + r, err := http.ReadRequest(bufio.NewReader(right)) + if err != nil { + return + } + _, _ = io.Copy(io.Discard, r.Body) + _ = r.Body.Close() + calls.Add(1) + time.Sleep(tc.delay) + body := `{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"counted once"}],"isError":false}}` + response := &http.Response{StatusCode: http.StatusOK, Proto: "HTTP/1.1", ProtoMajor: 1, ProtoMinor: 1, + Header: http.Header{"Content-Type": {"application/json"}}, Body: io.NopCloser(strings.NewReader(body)), + ContentLength: int64(len(body)), Close: true} + _ = response.Write(right) + }) + return left, nil + } + started := time.Now() + var err error + if tc.method == "model" { + request, _ := http.NewRequestWithContext(ctx, http.MethodPost, "http://127.0.0.1/responses", strings.NewReader("{}")) + response, requestErr := client.Do(request) + err = requestErr + if response != nil { + _ = response.Body.Close() + } + } else { + mcp := &acpMCPClient{url: "http://127.0.0.1/mcp", client: client} + _, err = mcp.call(ctx, tc.method, map[string]any{"name": "counted_action", "arguments": map[string]any{}}) + } + if (err == nil) != tc.wantResult || time.Since(started) != tc.wantTime || calls.Load() != 1 { + t.Errorf("held request: success=%v elapsed=%s calls=%d", err == nil, time.Since(started), calls.Load()) + } + if client.Timeout != 120*time.Second { + t.Error("tool wait changed the shared model/discovery client") + } + wg.Wait() + }) + }) + } +} + +func TestACPHeldApprovalsNeedSeparateDecisionsAndLeaveOtherSessionsAvailable(t *testing.T) { + var models, executions atomic.Int32 + pending := make(chan int, 2) + decisions := make(chan bool, 2) + mcp := &acpTestMCP{tools: func() []map[string]any { return acpTestTools("counted_action") }} + mcp.execute = func(w http.ResponseWriter, r *http.Request, id json.RawMessage, _ string, args json.RawMessage) { + var proposed struct { + Number int `json:"number"` + } + if json.Unmarshal(args, &proposed) != nil { + t.Error("invalid proposed action") + return + } + pending <- proposed.Number + select { + case <-r.Context().Done(): + return + case approved := <-decisions: + if !approved || r.Context().Err() != nil { + return + } + } + executions.Add(1) + acpTestToolResult(w, id, `{"executed":true}`, false) + } + peer := newACPTestPeer(t, foundry.ToolSchemaModeRequest, func(w http.ResponseWriter, r *http.Request) { + body := acpTestReadProvider(t, r) + switch models.Add(1) { + case 1: + acpTestCompleted(w, "first-proposal", "", acpTestCall("counted_action", "first-call", `{"number":1}`)) + case 2: + acpAssertApprovalContinuation(t, body, "first-proposal", "first-call", false) + acpTestCompleted(w, "second-proposal", "", acpTestCall("counted_action", "second-call", `{"number":2}`)) + case 3: + acpAssertApprovalContinuation(t, body, "second-proposal", "second-call", false) + acpTestCompleted(w, "final", "Both actions completed.") + default: + t.Error("model submission repeated") + } + }, mcp) + id := peer.prompt("Propose two counted actions.") + peer.read() // Consume the first tool-start event while its review stays held. + if <-pending != 1 || models.Load() != 1 || executions.Load() != 0 { + t.Fatal("action or continuation ran before its first decision") + } + other := newACPTestPeer(t, foundry.ToolSchemaModeRequest, func(w http.ResponseWriter, _ *http.Request) { + acpTestCompleted(w, "unrelated-response", "An independent session progressed.") + }, &acpTestMCP{}) + acpAssertStop(t, other.reply(other.prompt("Continue independently.")), "end_turn") + if models.Load() != 1 || executions.Load() != 0 { + t.Fatal("unrelated progress released the held action") + } + decisions <- true + peer.read() // First action completed. + peer.read() // Second action proposed. + if <-pending != 2 || executions.Load() != 1 || models.Load() != 2 { + t.Fatal("the first approval authorized a later action") + } + decisions <- true + acpAssertStop(t, peer.reply(id), "end_turn") + if executions.Load() != 2 || mcp.calls.Load() != 2 || models.Load() != 3 { + t.Fatal("approved actions did not execute and continue exactly once") + } +} + +func acpAssertApprovalContinuation(t *testing.T, body map[string]json.RawMessage, previous, call string, isError bool) { + t.Helper() + var outputs []foundry.FunctionOutput + var previousID string + if json.Unmarshal(body["previous_response_id"], &previousID) != nil || previousID != previous || + json.Unmarshal(body["input"], &outputs) != nil || len(outputs) != 1 || outputs[0].CallID != call { + t.Error("approval result did not continue its original response and pending call") + return + } + var result struct { + IsError bool `json:"isError"` + } + if json.Unmarshal([]byte(outputs[0].Output), &result) != nil || result.IsError != isError { + t.Error("approval result changed the execution outcome") + } +} + +func TestACPHeldApprovalFinalErrors(t *testing.T) { + for _, code := range []string{"approval_declined", "approval_expired", "approval_cancelled", "approval_stale", "tool_execution_failed", "tool_outcome_unknown"} { + t.Run(code, func(t *testing.T) { + var models, executions atomic.Int32 + pending, decision := make(chan struct{}), make(chan struct{}) + mcp := &acpTestMCP{tools: func() []map[string]any { return acpTestTools("counted_action") }} + mcp.execute = func(w http.ResponseWriter, r *http.Request, id json.RawMessage, _ string, _ json.RawMessage) { + close(pending) + select { + case <-r.Context().Done(): + return + case <-decision: + } + if code == "tool_execution_failed" || code == "tool_outcome_unknown" { + executions.Add(1) + } + acpTestMCPResult(w, id, map[string]any{ + "content": []map[string]string{{"type": "text", "text": "Safe final outcome."}}, "isError": true, + "structuredContent": map[string]any{"isError": true, "code": code}, + "_meta": map[string]any{"reviewer": "private-reviewer"}, + }) + } + peer := newACPTestPeer(t, foundry.ToolSchemaModeRequest, func(w http.ResponseWriter, r *http.Request) { + body := acpTestReadProvider(t, r) + if models.Add(1) == 1 { + acpTestCompleted(w, "held-response", "", acpTestCall("counted_action", "held-call", `{}`)) + return + } + acpAssertApprovalContinuation(t, body, "held-response", "held-call", true) + if !bytes.Contains(body["input"], []byte(code)) || bytes.Contains(body["input"], []byte("private-reviewer")) { + t.Error("final outcome lost its code or exposed review metadata") + } + acpTestCompleted(w, "final-response", "The final outcome was received.") + }, mcp) + id := peer.prompt("Propose a counted action.") + peer.read() + <-pending + if models.Load() != 1 || executions.Load() != 0 { + t.Fatal("pending approval executed or resumed the model") + } + close(decision) + acpAssertStop(t, peer.reply(id), "end_turn") + wantExecutions := int32(0) + if code == "tool_execution_failed" || code == "tool_outcome_unknown" { + wantExecutions = 1 + } + if models.Load() != 2 || executions.Load() != wantExecutions || mcp.calls.Load() != 1 { + t.Fatal("final error repeated an action or a model continuation") + } + }) + } +} + +func TestACPHeldApprovalCancellationAndLostResultNeverReplay(t *testing.T) { + for _, mode := range []string{"cancelled", "result_lost"} { + t.Run(mode, func(t *testing.T) { + var models, executions atomic.Int32 + pending, decision, disconnected := make(chan struct{}), make(chan struct{}), make(chan struct{}) + mcp := &acpTestMCP{tools: func() []map[string]any { return acpTestTools("counted_action") }} + mcp.execute = func(w http.ResponseWriter, r *http.Request, _ json.RawMessage, _ string, _ json.RawMessage) { + close(pending) + select { + case <-r.Context().Done(): + close(disconnected) + return + case <-decision: + } + executions.Add(1) + connection, _, err := w.(http.Hijacker).Hijack() + if err == nil { + _ = connection.Close() + } + } + peer := newACPTestPeer(t, foundry.ToolSchemaModeRequest, func(w http.ResponseWriter, _ *http.Request) { + models.Add(1) + acpTestCompleted(w, "held-response", "", acpTestCall("counted_action", "held-call", `{}`)) + }, mcp) + id := peer.prompt("Propose a counted action.") + peer.read() + <-pending + if mode == "cancelled" { + peer.send(map[string]any{"jsonrpc": "2.0", "method": "session/cancel", "params": map[string]string{"sessionId": peer.session}}) + acpAssertStop(t, peer.reply(id), "cancelled") + <-disconnected + close(decision) // A decision after cancellation has no waiting action. + } else { + close(decision) + acpAssertFailure(t, peer.reply(id)) + } + if peer.reply(peer.prompt("Try the same session again."))["error"] == nil { + t.Fatal("a poisoned approval session accepted another prompt") + } + wantExecutions := int32(0) + if mode == "result_lost" { + wantExecutions = 1 + } + if models.Load() != 1 || executions.Load() != wantExecutions || mcp.calls.Load() != 1 { + t.Fatal("cancelled or uncertain action was repeated or continued") + } + }) + } +} diff --git a/internal/acp/config.go b/internal/acp/config.go index 78a1949..5c222dc 100644 --- a/internal/acp/config.go +++ b/internal/acp/config.go @@ -22,6 +22,9 @@ const ( acpMaxMessageBytes = 8 << 20 acpTextChunkBytes = 32 << 10 acpHTTPTimeout = 120 * time.Second + // Orka permits 600s of review and 240s of approved execution. The remaining + // minute covers delivery; Task cancellation and lease loss can end it sooner. + acpToolCallTimeout = 900 * time.Second ) var ( @@ -135,6 +138,7 @@ func newACPHTTPClient() *http.Client { MaxIdleConnsPerHost: 2, MaxConnsPerHost: 2, IdleConnTimeout: 30 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, }, } } diff --git a/internal/acp/mcp.go b/internal/acp/mcp.go index d2e5141..7aefc94 100644 --- a/internal/acp/mcp.go +++ b/internal/acp/mcp.go @@ -9,6 +9,7 @@ import ( "net/http" "strings" "sync/atomic" + "time" "github.com/orka-agents/agent-runtime-foundry/internal/foundry" "github.com/orka-agents/agent-runtime-foundry/internal/strictjson" @@ -62,7 +63,7 @@ func (m *acpMCPClient) initialize(ctx context.Context) error { return errACPMCP } body, _ := json.Marshal(acpRequest{JSONRPC: "2.0", Method: "notifications/initialized"}) - response, err := m.post(ctx, body) + response, err := m.post(ctx, body, acpHTTPTimeout) if err != nil { return err } @@ -150,7 +151,11 @@ func (m *acpMCPClient) call(ctx context.Context, method string, params any) (jso return nil, errACPMCP } body, _ := json.Marshal(acpRequest{JSONRPC: "2.0", ID: id, Method: method, Params: encoded}) - response, err := m.post(ctx, body) + timeout := acpHTTPTimeout + if method == "tools/call" { + timeout = acpToolCallTimeout + } + response, err := m.post(ctx, body, timeout) if err != nil { return nil, err } @@ -176,7 +181,7 @@ func (m *acpMCPClient) call(ctx context.Context, method string, params any) (jso return reply.Result, nil } -func (m *acpMCPClient) post(ctx context.Context, body []byte) (*http.Response, error) { +func (m *acpMCPClient) post(ctx context.Context, body []byte, timeout time.Duration) (*http.Response, error) { request, err := http.NewRequestWithContext(ctx, http.MethodPost, m.url, bytes.NewReader(body)) if err != nil { return nil, errACPMCP @@ -185,7 +190,11 @@ func (m *acpMCPClient) post(ctx context.Context, body []byte) (*http.Response, e request.Header.Set("Content-Type", "application/json") request.Header.Set("Accept", "application/json, text/event-stream") request.Header.Set("MCP-Protocol-Version", acpMCPVersion) - response, err := m.client.Do(request) + // A held approval is still this one request. Share the bounded loopback + // transport, but never lengthen a concurrent model or discovery request. + client := *m.client + client.Timeout = timeout + response, err := client.Do(request) if err != nil { return nil, errACPMCP } diff --git a/internal/broker/agentkit.go b/internal/broker/agentkit.go index e1550db..1760983 100644 --- a/internal/broker/agentkit.go +++ b/internal/broker/agentkit.go @@ -81,8 +81,12 @@ func brokerAgentKitOutputs(request *foundry.ResponseRequest) error { if message.Len() == 0 { message.WriteString("The governed tool returned an error.") } + code, text := "brokered_tool_error", message.String() + if safeCode, safeText := brokerOrkaToolError(result.StructuredContent); safeCode != "" { + code, text = safeCode, safeText + } normalized = map[string]any{"approved": false, "error": map[string]string{ - "code": "brokered_tool_error", "message": message.String(), + "code": code, "message": text, }} } else { normalized = map[string]any{"approved": true, "output": result} @@ -95,3 +99,34 @@ func brokerAgentKitOutputs(request *foundry.ResponseRequest) error { } return nil } + +// Only Orka's structured final outcome selects an approval code. Tool text is +// not authority and may contain private decision details, so known outcomes +// always use a fixed model-visible message. None of these outcomes means wait. +func brokerOrkaToolError(raw json.RawMessage) (string, string) { + var outcome struct { + IsError bool `json:"isError"` + Code string `json:"code"` + } + if strictjson.DecodeStruct(raw, &outcome, false) != nil || !outcome.IsError { + return "", "" + } + var message string + switch outcome.Code { + case "approval_declined": + message = "The tool call was declined." + case "approval_expired": + message = "The tool approval expired." + case "approval_cancelled": + message = "The tool call was cancelled." + case "approval_stale": + message = "The tool approval is no longer valid." + case "tool_execution_failed": + message = "MCP tool execution failed." + case "tool_outcome_unknown": + message = "The tool execution outcome is unknown; do not retry." + default: + return "", "" + } + return outcome.Code, message +} diff --git a/internal/broker/agentkit_integration_test.go b/internal/broker/agentkit_integration_test.go index 30feecb..ac2c15f 100644 --- a/internal/broker/agentkit_integration_test.go +++ b/internal/broker/agentkit_integration_test.go @@ -34,9 +34,18 @@ func TestBrokerAgentKitHostedIntegration(t *testing.T) { if source == "" { t.Skip("set AGENTKIT_SOURCE_DIR and AGENTKIT_PYTHON to test an AgentKit checkout") } - for _, mode := range []string{"success", "tool_error", "authorization_denied", "gateway_strips_proof"} { + for _, mode := range []string{"success", "tool_error", "authorization_denied", "gateway_strips_proof", "approval_approved", "approval_declined", "approval_expired", "approval_execution_failed"} { t.Run(mode, func(t *testing.T) { - var models, calls, invocations atomic.Int32 + var models, calls, invocations, executions atomic.Int32 + pendingReview, decision, reviewChecked := make(chan struct{}), make(chan struct{}), make(chan struct{}) + heldApproval := strings.HasPrefix(mode, "approval_") + approvalCode := "" + switch mode { + case "approval_declined", "approval_expired": + approvalCode = mode + case "approval_execution_failed": + approvalCode = "tool_execution_failed" + } model := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, _ := io.ReadAll(r.Body) if r.Method != http.MethodPost || r.URL.Path != "/v1/chat/completions" || @@ -63,10 +72,14 @@ func TestBrokerAgentKitHostedIntegration(t *testing.T) { Output json.RawMessage `json:"output"` Error map[string]any `json:"error"` } - approved := mode != "tool_error" || n != 2 + approved := n != 2 || (mode != "tool_error" && approvalCode == "") + code := "brokered_tool_error" + if n == 2 && approvalCode != "" { + code = approvalCode + } if last["role"] != "tool" || json.Unmarshal([]byte(text), &result) != nil || result.Approved == nil || *result.Approved != approved || - (!approved && (result.Output != nil || result.Error["code"] != "brokered_tool_error")) { + (!approved && (result.Output != nil || result.Error["code"] != code)) || strings.Contains(text, "private-review-") { t.Error("real model resume lost the governed tool result or converted an error to approval") w.WriteHeader(http.StatusBadRequest) return @@ -191,10 +204,25 @@ func TestBrokerAgentKitHostedIntegration(t *testing.T) { w.WriteHeader(http.StatusBadRequest) return } - failed := mode == "tool_error" && n == 1 + if n == 1 && heldApproval { + close(pendingReview) + select { + case <-decision: + case <-r.Context().Done(): + return + } + } + if n != 1 || approvalCode == "" || approvalCode == "tool_execution_failed" { + executions.Add(1) + } + failed := n == 1 && (mode == "tool_error" || approvalCode != "") structured := map[string]any{"part": "FBR-7", "quantity": 7} if failed { structured = map[string]any{"isError": true, "error": "MCP tool execution failed"} + if approvalCode != "" { + structured["code"] = approvalCode + structured["error"] = "private-review-note" + } } text, _ := json.Marshal(structured) result = map[string]any{"content": []map[string]string{{"type": "text", "text": string(text)}}, @@ -221,6 +249,28 @@ func TestBrokerAgentKitHostedIntegration(t *testing.T) { if result["sessionId"] == nil { t.Fatal("real ACP session creation failed") } + if heldApproval { + go func() { + defer close(reviewChecked) + select { + case <-pendingReview: + case <-t.Context().Done(): + return + } + if models.Load() != 1 || calls.Load() != 1 || executions.Load() != 0 || invocations.Load() != 1 { + t.Error("held review executed a tool or continued the real hosted model") + } + b.mu.Lock() + owner := b.ledger.Sessions[foundry.JSONDigest(c.Owner)] + prompt := owner.Prompts[c.promptKey()] + owned := !prompt.Closing && prompt.LastSequence == 1 && len(owner.Responses[prompt.LastAlias].CallIDs) == 1 + b.mu.Unlock() + if !owned { + t.Error("held review lost its durable original call ownership") + } + close(decision) + }() + } terminal := peer.call("session/prompt", map[string]any{"sessionId": result["sessionId"], "prompt": []any{map[string]string{"type": "text", "text": "Check the work order, then check inventory."}}}) blocked := mode == "authorization_denied" || mode == "gateway_strips_proof" @@ -240,6 +290,16 @@ func TestBrokerAgentKitHostedIntegration(t *testing.T) { } remoteMu.Unlock() } + if heldApproval { + <-reviewChecked + wantExecutions := int32(2) + if approvalCode == "approval_declined" || approvalCode == "approval_expired" { + wantExecutions = 1 + } + if executions.Load() != wantExecutions { + t.Fatal("human decision did not control the counted execution") + } + } _ = brokerTestControl(t, brokerServer.URL, brokerapi.SettlePath, c) _ = brokerTestControl(t, brokerServer.URL, brokerapi.RetirePath, c) for _, path := range []string{stateFile, filepath.Join(cfg.stateDir, "state.json")} { diff --git a/internal/broker/agentkit_test.go b/internal/broker/agentkit_test.go index d540c28..bceaa81 100644 --- a/internal/broker/agentkit_test.go +++ b/internal/broker/agentkit_test.go @@ -28,16 +28,34 @@ func brokerAgentKitFunctionBody(previous, call, output string) []byte { return body } +func brokerAgentKitOrkaError(code string) string { + raw, _ := json.Marshal(map[string]any{ + "content": []map[string]string{{"type": "text", "text": "private-review-note"}}, "isError": true, + "structuredContent": map[string]any{"isError": true, "code": code, "reviewer": "private-reviewer"}, + }) + return string(raw) +} + func TestBrokerAgentKitContinuationWireAndIsolation(t *testing.T) { for _, tc := range []struct { - name string - output string - approved bool + name string + output string + approved bool + errorCode string }{ - {"text", `{"content":[{"type":"text","text":"fixture tool result"}],"isError":false}`, true}, - {"structured", `{"content":[{"type":"text","text":"{\"count\":7}"}],"isError":false,"structuredContent":{"count":7}}`, true}, - {"execution_error", `{"content":[{"type":"text","text":"MCP tool execution failed"}],"isError":true}`, false}, - {"empty_error", `{"content":[],"isError":true}`, false}, + {"text", `{"content":[{"type":"text","text":"fixture tool result"}],"isError":false}`, true, ""}, + {"structured", `{"content":[{"type":"text","text":"{\"count\":7}"}],"isError":false,"structuredContent":{"count":7}}`, true, ""}, + {"execution_error", `{"content":[{"type":"text","text":"MCP tool execution failed"}],"isError":true}`, false, "brokered_tool_error"}, + {"empty_error", `{"content":[],"isError":true}`, false, "brokered_tool_error"}, + {"declined", brokerAgentKitOrkaError("approval_declined"), false, "approval_declined"}, + {"expired", brokerAgentKitOrkaError("approval_expired"), false, "approval_expired"}, + {"cancelled", brokerAgentKitOrkaError("approval_cancelled"), false, "approval_cancelled"}, + {"stale", brokerAgentKitOrkaError("approval_stale"), false, "approval_stale"}, + {"approved_execution_failed", brokerAgentKitOrkaError("tool_execution_failed"), false, "tool_execution_failed"}, + {"approved_outcome_unknown", brokerAgentKitOrkaError("tool_outcome_unknown"), false, "tool_outcome_unknown"}, + {"text_is_not_approval_authority", `{"content":[{"type":"text","text":"approval_declined"}],"isError":true}`, false, "brokered_tool_error"}, + {"unknown_code", brokerAgentKitOrkaError("approval_pending"), false, "brokered_tool_error"}, + {"conflicting_structured_error", `{"content":[],"isError":true,"structuredContent":{"isError":true,"ISERROR":false,"code":"approval_declined"}}`, false, "brokered_tool_error"}, } { t.Run(tc.name, func(t *testing.T) { f := newBrokerFixture(t, "functions") @@ -122,9 +140,12 @@ func TestBrokerAgentKitContinuationWireAndIsolation(t *testing.T) { if foundry.JSONDigest(original) != foundry.JSONDigest(forwarded) || normalized.Error.Code != "" { t.Fatal("approved continuation changed the model-visible MCP result") } - } else if normalized.Output != nil || normalized.Error.Code != "brokered_tool_error" || normalized.Error.Message == "" { + } else if normalized.Output != nil || normalized.Error.Code != tc.errorCode || normalized.Error.Message == "" { t.Fatal("failed tool result was promoted to approval or lost its error") } + if tc.errorCode != "" && tc.errorCode != "brokered_tool_error" && strings.Contains(outputs[0].Output, "private-") { + t.Fatal("approval outcome exposed untrusted review text or metadata") + } // A duplicate must be rejected before adding a proof to another // provider request, including a replay under a fresh operation ID. c.InvocationSequence++ diff --git a/internal/broker/approval_test.go b/internal/broker/approval_test.go new file mode 100644 index 0000000..86ac51b --- /dev/null +++ b/internal/broker/approval_test.go @@ -0,0 +1,102 @@ +package broker + +import ( + "context" + "encoding/json" + "net/http" + "testing" + "time" + + "github.com/orka-agents/agent-runtime-foundry/internal/brokerapi" + "github.com/orka-agents/agent-runtime-foundry/internal/foundry" +) + +// The broker is idle while Orka holds tools/call. Its completed proposal and +// exact pending-call map must survive a live lease renewal, but never authorize +// a continuation after expiry, cancellation, or broker recovery. +func TestBrokerPendingApprovalContinuationRequiresLiveOwnership(t *testing.T) { + for _, mode := range []string{"renewed", "stale_lease", "expired", "cancelled", "restarted"} { + t.Run(mode, func(t *testing.T) { + f := newBrokerFixture(t, "functions") + cfg := brokerTestConfig(t, f) + cfg.agentKitProof = brokerAgentKitFixtureProof + b, server := startBrokerTest(t, cfg) + c := brokerTestContext(cfg) + if mode == "renewed" || mode == "expired" { + c.LeaseExpiresAt = time.Now().Add(time.Second).UTC().Format(time.RFC3339Nano) + } + status, raw, err := brokerTestHTTP(context.Background(), server.URL, brokerapi.ResponsesPath, c, brokerTestBody("")) + var proposal foundry.Response + if err != nil || status != http.StatusOK || json.Unmarshal(raw, &proposal) != nil || len(proposal.Output) != 1 { + t.Fatal("initial pending tool call was not durably recorded") + } + original := c + switch mode { + case "renewed", "stale_lease": + renewal := c + renewal.LeaseGeneration++ + renewal.LeaseExpiresAt = time.Now().Add(2 * time.Minute).UTC().Format(time.RFC3339Nano) + proof := brokerTestControl(t, server.URL, brokerapi.RenewPath, renewal) + if proof.State != "open" || proof.LeaseGeneration != 2 { + t.Fatal("idle approval wait did not retain its renewed ownership") + } + if mode == "renewed" { + c = renewal + } + case "cancelled": + _ = brokerTestControl(t, server.URL, brokerapi.SettlePath, c) + case "restarted": + b.close() + server.Close() + b, server = startBrokerTest(t, cfg) + } + if mode == "renewed" || mode == "expired" { + expiry, _ := time.Parse(time.RFC3339Nano, original.LeaseExpiresAt) + time.Sleep(time.Until(expiry.Add(50 * time.Millisecond))) + } + _, inferences, _, _ := f.counts() + if inferences != 1 { + t.Fatal("a pending review submitted model work") + } + c.InvocationSequence++ + c.OperationID = "held-approval-result" + body := brokerAgentKitFunctionBody(proposal.ID, proposal.Output[0].CallID, brokerAgentKitOrkaError("approval_declined")) + status, _, err = brokerTestHTTP(context.Background(), server.URL, brokerapi.ResponsesPath, c, body) + wantStatus := http.StatusGone + if mode == "renewed" { + wantStatus = http.StatusOK + } else if mode == "stale_lease" { + wantStatus = http.StatusConflict + } + if err != nil || status != wantStatus { + t.Fatalf("held continuation status=%d, want %d", status, wantStatus) + } + c.InvocationSequence++ + c.OperationID = "repeated-held-approval-result" + status, _, err = brokerTestHTTP(context.Background(), server.URL, brokerapi.ResponsesPath, c, body) + if err != nil || status == http.StatusOK { + t.Fatal("a repeated or stale decision resubmitted the original tool result") + } + _, inferences, _, _ = f.counts() + wantInferences := 1 + if mode == "renewed" { + wantInferences = 2 + } + if inferences != wantInferences { + t.Fatal("lost ownership or duplicate continuation repeated model work") + } + if mode == "restarted" { + b.mu.Lock() + session := b.ledger.Sessions[foundry.JSONDigest(c.Owner)] + stored := session.Responses[proposal.ID] + preserved := session.Prompts[c.promptKey()].Closing && stored.CallIDs[proposal.Output[0].CallID] != "" + b.mu.Unlock() + if !preserved { + t.Fatal("broker restart erased the closed original approval ownership") + } + } + _ = brokerTestControl(t, server.URL, brokerapi.SettlePath, c) + _ = brokerTestControl(t, server.URL, brokerapi.RetirePath, c) + }) + } +}