From 11b4c4019351d8e35f873db97d87b3dfd74c03f2 Mon Sep 17 00:00:00 2001 From: Qiyuanqiii <2297740147@qq.com> Date: Mon, 5 Oct 2026 15:29:32 +0800 Subject: [PATCH] fix: complete MCP execution through CLI and Thread approvals Connect persisted Workspace and Session bindings, use current Gateway read arguments, declare an object root for dynamic MCP schemas, and stop Thread handoffs while waiting for approval. Record real model, Skill, MCP and cancellation acceptance plus the frontend handoff. Co-authored-by: Codex --- .../2026-10-05-core-integrations.md | 130 ++++++++++++++++++ docs/architecture.md | 2 +- internal/app/mcp_command.go | 6 +- internal/app/mcp_command_test.go | 93 +++++++++++++ .../mcp_approval_recovery_acceptance_test.go | 8 +- .../supervisor_tool_permission_test.go | 6 + internal/application/supervisor_tools.go | 3 +- .../application/thread_mcp_approval_test.go | 68 +++++++++ internal/application/thread_turn.go | 6 + internal/mcp/tools.go | 36 ++++- 10 files changed, 350 insertions(+), 8 deletions(-) create mode 100644 docs/acceptance/2026-10-05-core-integrations.md create mode 100644 internal/app/mcp_command_test.go create mode 100644 internal/application/thread_mcp_approval_test.go diff --git a/docs/acceptance/2026-10-05-core-integrations.md b/docs/acceptance/2026-10-05-core-integrations.md new file mode 100644 index 00000000..688a1a5b --- /dev/null +++ b/docs/acceptance/2026-10-05-core-integrations.md @@ -0,0 +1,130 @@ +# Core integration acceptance — 2026-10-05 + +This acceptance covers representative product entry points before the next +frontend iteration. The architecture cleanup through PR #252 is merged in main +`9559e09`. Acceptance used its implementation at `14b5fc02`, plus the MCP fixes +described below, on Windows with isolated `CYBERAGENT_HOME` directories. +The user's normal database was not used. + +**Decision:** the core backend is ready for frontend integration with this +patch. The remaining native ZIP and CLI approval entry gaps below have explicit +supported alternatives and do not block the Thread interface. + +## Results + +| Path | Evidence and result | +| --- | --- | +| DeepSeek model qualification | The real `deepseek-v4-flash` endpoint passed the native tool protocol qualification through the existing system credential. Two model requests, one synthetic tool request, no actual tool execution during qualification. | +| HTTP Thread → model → native Skill | A bounded interactive Code Run completed a real HTTP turn. The model discovered the installed Skill, called `skill_read` for its instructions and then `references/acceptance.txt`, and returned both markers that were absent from the user prompt. Both calls are `completed` in the durable tool ledger. | +| Model transport retry | The first request in that turn encountered a network failure before any stream bytes. The existing retry completed successfully. The transcript preserves the failed attempt and the final successful response. | +| Interrupt and request replay | After a model call became active, Thread interrupt returned 202 / `stopping`; the turn returned 499 / `CANCELLED` with durable `turn_failure`. Repeating the original request key returned the same failure without another model attempt. | +| API restart | After stopping and restarting the API on the same isolated home, messages, transcript, Run history and request observations remained readable. The successful request was `completed`; the canceled request was `failed`, with `last_turn_interrupted: true` and execution `idle`. | +| Skill installation lifecycle | CLI directory import, exact replay, cross-process read and changed-input rejection passed. HTTP stage, approval, enable, disable, stale-generation rejection, re-enable and revoke passed. Disabled state survived an API restart; enabling a revoked installation returned 412. | +| HTTP access and request validation | Read-token mutation returned 401. Unknown Thread creation fields and unsupported versions returned 400; these rejected requests created no Thread. | +| Real stdio MCP server | After the fix, initialization, tool/resource discovery, file read and directory list passed. A path outside the Workspace, an extra Workspace-root field and a shell tool request were rejected. Closing stdin exited the process; a new process could read the same persisted Run. | +| MCP client lifecycle | Actual CLI stage, review, capability refresh and enable passed. Starting a missing server failed as expected; explicit refresh recovered the server after its launcher was restored. The peer was the product MCP server, not a fabricated protocol response. | +| HTTP Thread → model → approval → MCP | With both entry fixes, a fresh bounded Run returned 202 with one pending MCP call. Preview named the exact read-only operation; `approve_once` returned 202 with completed continuation. One completed MCP receipt and the model's final artifact digest were recorded. | + +The Skill turn used two tools and one completed agent turn. Reported model usage +was 16,815 tokens. Four model attempts include one network failure with unknown +usage; the known token total is not a complete billing statement. Monetary +tracking was disabled; explicit turn, tool, token and timeout budgets bounded +the acceptance Runs. + +## MCP defects found through product entry points + +The original `mcp serve` process initialized and advertised tools but could not +execute `read_file` or `list_workspace`. Three entry-point mismatches caused this: + +1. The CLI constructed a Gateway without its persisted Workspace root resolver. +2. The server sent read arguments through `Payload` and supplied execution + identifiers that the current Gateway owns. The Gateway expects read tools in + `Arguments` and rejects those caller-supplied identifiers. +3. The server omitted the Run's Session identity, preventing successful tool + output from being recorded as an artifact. + +The patch connects the persisted Workspace resolver and adapts the server's +existing read-only tool contract to the current Gateway. It does not add tools +or new execution authority. A subprocess regression exercises the actual +`Execute → mcp serve → SQLite → Gateway` chain from a working directory different +from the selected Workspace. + +The first real model request with the dynamic MCP catalog also returned HTTP +400. Its schema root contained only `oneOf`; the patch adds `type: object` +while retaining every exact server/tool/fingerprint branch. This matches the +object input schema in the [Messages API reference](https://platform.claude.com/docs/en/api/messages/create). +With that change, the same DeepSeek integration produced the expected MCP call. +HTTP `approve_once` resumed it, recorded a completed MCP client receipt and +returned the file artifact's digest to the model. + +That call exposed a second Thread-entry bug: creating an MCP approval stopped +the handoff at `root_wait`, but the durable Run stayed `running`. The Thread +service opened another handoff while the same input was still pending, producing +HTTP 500 instead of returning the approval boundary. The patch stops the Thread +at `root_wait`; approval decision and continuation retain their existing owners. +The final real HTTP acceptance returned 202 at both the pending and approval +steps and committed the original message after one tool execution. + +The focused subprocess test and existing MCP protocol/interoperability tests +passed, as did the CLI build: + +```text +go test ./internal/app -run '^TestMCPServeCLIUsesPersistedWorkspaceForReadTools$' -count=1 +go test ./internal/mcp -run '^(TestMCP.*|TestSDKLegacyClientInteroperatesWithNativeServer)$' -count=1 +go test ./internal/application -run '^TestSupervisorMCPRequiresReviewedSnapshotAndExactRuntimeScope$' -count=1 +go test ./internal/application -run '^TestThreadTurnStopsAtPendingMCPApprovalAndResumesSameTurn$' -count=1 +go build ./cmd/cyberagent +``` + +## Frontend handoff + +- Use the existing Thread API for the conversation. `Run.status: running` + together with `Thread execution.state: idle` means the interactive Run is + ready for another message; it does not mean the model is still generating. +- Merge transcript stages by `canonical_id` / `durable_call_id`. A single tool + can have pending, running, completed and result-recorded events. Count calls + using the Run's `tool_usage`, not the number of completed transcript events. +- Render the settled turn result after a successful retry. A historical failed + model attempt must not overwrite the final reply. +- Interrupt using the current `execution_id`. Inspect a submitted request with + `GET /threads/{id}/turn-request` and its original `Idempotency-Key`. A sealed + failure is not retried by resending that key; a new user intent needs a new key. +- Use `GET /runs/{id}` for checkpoint token totals and `tool_usage`. + `/runs/{id}/usage` is not an HTTP endpoint. The richer `run usage` view is + currently CLI-only. HTTP Thread creation currently uses default budgets and + does not accept custom budget fields. +- Skill enablement depends on installation `state` as well as capabilities. + Disabling retains the approved capability set. Staging is not activation; + the interface needs separate staged, approved, enabled, disabled and revoked + states. +- Use the HTTP approval preview and decision endpoints for MCP calls in Ask + mode. `approve_once` resumes the original pending call; do not submit a new + turn to repeat it. The CLI approval continuation currently only supports Web + Fetch and cannot complete an MCP Ask approval by itself. + +## Explicit remaining product gap + +The current Web ZIP upload uses the legacy package request without a native +snapshot. Uploading an ordinary native `SKILL.md` ZIP through this entry returns +400 `INVALID_ARGUMENT`, even though CLI native-directory import and HTTP import +with a Go-produced snapshot work. The current success copy also says installed +when the result can be only staged. + +Do not advertise ordinary native ZIP upload as supported. The frontend can use +the existing Desktop native-directory acquisition path, or keep the Web upload +explicitly limited to legacy packages. Supporting native ZIP directly needs a +Go-owned archive acquisition step before staging; the renderer should not invent +the snapshot. This is a scoped follow-up, not a reason to continue architecture +cleanup before building the Thread interface. + +## Evidence + +Local acceptance records are under +`D:/CodexAcceptance/universal-code-20261005/`, grouped by `root`, `model`, +`skills` and `mcp`. They include actual HTTP responses, CLI outputs, stdio MCP +transcripts and the two completed Skill tool records. API credentials and the +isolated databases are local only and are not repository artifacts. + +Representative checks cover the installed DeepSeek provider, a native Skill +fixture and the product's stdio MCP implementation. They do not certify every +third-party provider, MCP transport, extension format or Desktop interaction. diff --git a/docs/architecture.md b/docs/architecture.md index bee528d2..9dda416c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -30,7 +30,7 @@ Implementation status for this revision: | CI execution | Deduplicated | [CI](../.github/workflows/ci.yml) runs analyzer vectors within the Go suite and desktop boundary tests through the platform build scripts. Platform, race and standalone release checks remain distinct. | | Model and extension integrations | Existing boundaries | `llm` owns provider protocols; MCP shares its SDK execution path. Skills and Plugins use their supported formats. Compatibility is tracked per integration. | | Command runtime construction and teardown | Shared for Desktop and HTTP | [Command runtime set](../internal/application/command_runtime_set.go) assembles Host, Local and Docker adapters and shuts down their managers. Each entry selects its enabled backends; CLI retains its invocation-owned lifetime. | -| Ecosystem product validation | Follow-up | Exercise representative integrations through install, configuration, authentication, execution, failure and cancellation. External Agent orchestration remains outside this phase. | +| Ecosystem product validation | Representative paths accepted | [2026-10-05 acceptance](acceptance/2026-10-05-core-integrations.md) covers real DeepSeek/Thread/Skill execution, cancellation and restart, Skill lifecycle and stdio MCP entry points. Ordinary native Skill ZIP upload remains a scoped product gap; third-party compatibility stays integration-specific. | The Gateway extension target is concrete: add a tool implementation and one registration; Gateway discovery and dispatch derive from that entry. A new tool diff --git a/internal/app/mcp_command.go b/internal/app/mcp_command.go index 1a6f3866..00cfd81a 100644 --- a/internal/app/mcp_command.go +++ b/internal/app/mcp_command.go @@ -53,7 +53,11 @@ func (a *App) mcpServeCommand(ctx context.Context, args []string) error { if fs.NArg() != 0 || strings.TrimSpace(*runID) == "" || strings.TrimSpace(*workspaceID) == "" { return errors.New("usage: cyberagent mcp serve --run --workspace ") } - gateway := toolgateway.New(a.store, a.checker) + gateway := toolgateway.New(a.store, a.checker). + WithWorkspaceRootResolver(func(ctx context.Context, workspaceID string) (string, error) { + record, err := a.store.GetWorkspaceByID(ctx, workspaceID) + return record.RootPath, err + }) server, err := mcp.New(mcp.Options{Store: a.store, Tools: gateway, RunID: *runID, WorkspaceID: *workspaceID, SessionTTL: *sessionTTL, CallTimeout: *callTimeout, MaxConcurrent: *maxConcurrent}) diff --git a/internal/app/mcp_command_test.go b/internal/app/mcp_command_test.go new file mode 100644 index 00000000..211c8c5b --- /dev/null +++ b/internal/app/mcp_command_test.go @@ -0,0 +1,93 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "cyberagent-workbench/internal/mcp" +) + +func TestMCPServeCLIUsesPersistedWorkspaceForReadTools(t *testing.T) { + t.Setenv("CYBERAGENT_HOME", t.TempDir()) + command := func(args ...string) string { + t.Helper() + stdout, stderr, code := executeTestCommand(t, args...) + if code != 0 || stderr != "" { + t.Fatalf("command %v: code=%d stdout=%s stderr=%s", args, code, stdout, stderr) + } + return stdout + } + command("workspace", "init", "mcp-read") + shown := command("workspace", "show", "mcp-read") + workspaceID := fieldLine(shown, "id") + workspaceRoot := fieldLine(shown, "path") + if err := os.WriteFile(filepath.Join(workspaceRoot, "marker.txt"), []byte("workspace marker\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(filepath.Dir(workspaceRoot), "outside.txt"), []byte("outside marker\n"), 0o600); err != nil { + t.Fatal(err) + } + created := command("run", "create", "Read the MCP workspace", "--workspace", "mcp-read") + runID := strings.Fields(created)[1] + executable, err := os.Executable() + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + process := exec.CommandContext(ctx, executable, "-test.run=^TestMCPServeCLIHelper$", "--", runID, workspaceID) + // The process working directory is deliberately not the Workspace root. + process.Dir = t.TempDir() + process.Env = append(os.Environ(), "CYBERAGENT_MCP_CLI_HELPER=1") + process.Stdin = strings.NewReader(strings.Join([]string{ + `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"cli-test","version":"1"}}}`, + `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"marker.txt"}}}`, + `{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"list_workspace","arguments":{"path":"."}}}`, + `{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"../outside.txt"}}}`, + `{"jsonrpc":"2.0","id":5,"method":"tools/call","params":{"name":"shell","arguments":{"command":"echo forbidden"}}}`, + `{"jsonrpc":"2.0","id":6,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"marker.txt","workspace_root":"outside"}}}`, + }, "\n") + "\n") + var stderr bytes.Buffer + process.Stderr = &stderr + stdout, err := process.Output() + if err != nil || stderr.Len() != 0 { + t.Fatalf("MCP CLI process: %v stderr=%s stdout=%s", err, stderr.String(), stdout) + } + responses := make(map[string]mcp.Envelope) + for _, line := range strings.Split(strings.TrimSpace(string(stdout)), "\n") { + var envelope mcp.Envelope + if err := json.Unmarshal([]byte(line), &envelope); err != nil { + t.Fatalf("stdout is not MCP JSON: %v: %s", err, line) + } + responses[string(envelope.ID)] = envelope + } + for _, id := range []string{"2", "3"} { + var result mcp.CallToolResult + if err := json.Unmarshal(responses[id].Result, &result); err != nil || len(result.Content) != 1 || + !strings.Contains(result.Content[0].Text, `"status":"completed"`) { + encoded, _ := json.Marshal(responses[id]) + t.Fatalf("workspace read %s did not complete: %s", id, encoded) + } + } + if responses["4"].Error == nil || responses["5"].Error == nil || responses["5"].Error.Code != mcp.CodeMethodNotFound { + t.Fatalf("MCP read scope widened: traversal=%+v shell=%+v", responses["4"], responses["5"]) + } + if responses["6"].Error == nil || responses["6"].Error.Code != mcp.CodeInvalidParams { + t.Fatalf("schema-external workspace root was accepted: %+v", responses["6"]) + } +} + +func TestMCPServeCLIHelper(t *testing.T) { + if os.Getenv("CYBERAGENT_MCP_CLI_HELPER") != "1" { + t.Skip("MCP CLI subprocess only") + } + args := os.Args[len(os.Args)-2:] + os.Exit(Execute([]string{"mcp", "serve", "--run", args[0], "--workspace", args[1]}, os.Stdout, os.Stderr)) +} diff --git a/internal/application/mcp_approval_recovery_acceptance_test.go b/internal/application/mcp_approval_recovery_acceptance_test.go index 158f54a2..6e1f6510 100644 --- a/internal/application/mcp_approval_recovery_acceptance_test.go +++ b/internal/application/mcp_approval_recovery_acceptance_test.go @@ -162,7 +162,7 @@ func (f *mcpRecoveryAcceptanceFixture) unblock() { } } -func newMCPRecoveryAcceptanceFixture(t *testing.T, mode domain.RunExecutionPermissionMode, callCount int, blocked bool) *mcpRecoveryAcceptanceFixture { +func newMCPRecoveryAcceptanceRuntime(t *testing.T, mode domain.RunExecutionPermissionMode, callCount int, blocked bool) *mcpRecoveryAcceptanceFixture { t.Helper() f := &mcpRecoveryAcceptanceFixture{path: filepath.Join(t.TempDir(), "approval-recovery.db"), entered: make(chan struct{})} if blocked { @@ -228,6 +228,12 @@ func newMCPRecoveryAcceptanceFixture(t *testing.T, mode domain.RunExecutionPermi t.Fatal(err) } f.runtime.Store(true) + return f +} + +func newMCPRecoveryAcceptanceFixture(t *testing.T, mode domain.RunExecutionPermissionMode, callCount int, blocked bool) *mcpRecoveryAcceptanceFixture { + t.Helper() + f := newMCPRecoveryAcceptanceRuntime(t, mode, callCount, blocked) result, err := f.supervisor.Step(t.Context(), f.run.ID) if err != nil || result.RunStatus != domain.RunWaitingApproval || f.requests.Load() != 0 { t.Fatalf("initial consent boundary: result=%+v requests=%d err=%v", result, f.requests.Load(), err) diff --git a/internal/application/supervisor_tool_permission_test.go b/internal/application/supervisor_tool_permission_test.go index 55aa7867..61c1e91d 100644 --- a/internal/application/supervisor_tool_permission_test.go +++ b/internal/application/supervisor_tool_permission_test.go @@ -239,6 +239,12 @@ func TestSupervisorMCPRequiresReviewedSnapshotAndExactRuntimeScope(t *testing.T) !strings.Contains(string(schema), `"const":"`+fingerprint+`"`) { t.Fatalf("reviewed MCP capability was not encoded exactly: %s", schema) } + var schemaRoot struct { + Type string `json:"type"` + } + if err := json.Unmarshal(schema, &schemaRoot); err != nil || schemaRoot.Type != "object" { + t.Fatalf("MCP input schema must declare an object at its root: %s", schema) + } if found, _ := visible(domain.ExecutionSurfaceCode, domain.ExecutionPhaseDeliver, domain.RunExecutionPermissionAuto, options); !found { t.Fatal("Auto did not expose the reviewed MCP capability") diff --git a/internal/application/supervisor_tools.go b/internal/application/supervisor_tools.go index eed369ed..6050c3b7 100644 --- a/internal/application/supervisor_tools.go +++ b/internal/application/supervisor_tools.go @@ -712,8 +712,9 @@ func supervisorMCPToolSchema(capabilities mcp.ScopedCapabilities) json.RawMessag } } raw, err := json.Marshal(struct { + Type string `json:"type"` OneOf []choice `json:"oneOf"` - }{OneOf: choices}) + }{Type: "object", OneOf: choices}) if err != nil || len(raw) > maxSupervisorMCPSchemaBytes { return toolgateway.MCPToolDefinition().InputSchema } diff --git a/internal/application/thread_mcp_approval_test.go b/internal/application/thread_mcp_approval_test.go new file mode 100644 index 00000000..cdcaa261 --- /dev/null +++ b/internal/application/thread_mcp_approval_test.go @@ -0,0 +1,68 @@ +package application + +import ( + "testing" + + "cyberagent-workbench/internal/approval" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/runmutation" +) + +func TestThreadTurnStopsAtPendingMCPApprovalAndResumesSameTurn(t *testing.T) { + f := newMCPRecoveryAcceptanceRuntime(t, domain.RunExecutionPermissionAsk, 1, false) + request := ExecuteThreadTurnRequest{Version: domain.ThreadMessageProtocolVersion, + ThreadID: domain.InitialThreadID(f.run.ID), Content: "Look up the exact reviewed MCP input.", + OperationKey: "thread-mcp-pending-operation-0001", RequestedBy: "operator"} + result, err := f.turns.Execute(t.Context(), request) + if err != nil || result.Execution == nil || result.Execution.Handoff.Result == nil { + t.Fatalf("Thread did not return its approval boundary: result=%+v err=%v", result, err) + } + handoff := result.Execution.Handoff + if handoff.Result.Status != domain.RunExecutionHandoffCompleted || handoff.Result.StopReason != "root_wait" || + result.Submission.Message.Status != domain.OperatorSteeringPending || result.Submission.Run.Status != domain.RunRunning { + t.Fatalf("approval wait lost its resumable message or Run: %+v", result) + } + if f.requests.Load() != 0 || f.wireCalls.Load() != 0 { + t.Fatal("Thread dispatched the MCP call before operator consent") + } + var found bool + f.pending, found, err = f.st.GetSupervisorCheckpoint(t.Context(), f.run.ID) + if err != nil || !found || f.pending.Phase != domain.SupervisorTurnStarted { + t.Fatalf("missing original pending turn: %+v err=%v", f.pending, err) + } + rounds, err := f.st.ListSupervisorToolRounds(t.Context(), f.pending) + if err != nil || len(rounds) != 1 || len(rounds[0].Calls) != 1 { + t.Fatalf("missing exact durable MCP intent: %+v err=%v", rounds, err) + } + f.calls = rounds[0].Calls + f.assertApproval(t, 0, approval.StatusPending) + if retry, err := f.turns.Execute(t.Context(), request); err != nil || !retry.Replayed || retry.Submission.Message.ID != result.Submission.Message.ID { + t.Fatalf("pending request retry restarted or lost the original message: %+v err=%v", retry, err) + } + second := threadTurnExecutionRequest(SubmitThreadMessageRequest{ThreadID: request.ThreadID, OperationKey: request.OperationKey, + RequestedBy: request.RequestedBy}, f.run.ID, 2) + if _, found, err := f.st.GetRunExecutionHandoff(t.Context(), + runmutation.RunExecutionHandoffOperationDigest(f.run.ID, second.OperationKey)); err != nil || found { + t.Fatalf("Thread created a second handoff while waiting for approval: found=%t err=%v", found, err) + } + f.provider.mu.Lock() + modelRequests := f.provider.requests + f.provider.mu.Unlock() + if modelRequests != 1 { + t.Fatalf("approval wait repeated the initial model request: %d", modelRequests) + } + f.decide(t, f.request(t, 0, ApprovalControlApproveOnce, "thread-mcp-consent-operation-0001")) + if resumed := f.resume(t, 0); resumed.State != "completed" { + t.Fatalf("exact approved Thread turn did not resume: %+v", resumed) + } + f.assertCall(t, 0, domain.SupervisorToolCompleted, "", true) + f.assertTurnSettled(t, 1, 1) + message, err := f.st.GetOperatorSteering(t.Context(), result.Submission.Message.ID) + if err != nil || message.Status != domain.OperatorSteeringCommitted { + t.Fatalf("approval continuation did not commit the original Thread input: %+v err=%v", message, err) + } + if replay, err := f.turns.Execute(t.Context(), request); err != nil || !replay.Replayed || replay.Submission.Message.ID != message.ID { + t.Fatalf("settled Thread request did not replay: %+v err=%v", replay, err) + } + f.assertTurnSettled(t, 1, 1) +} diff --git a/internal/application/thread_turn.go b/internal/application/thread_turn.go index 5bd3a373..5cdfd786 100644 --- a/internal/application/thread_turn.go +++ b/internal/application/thread_turn.go @@ -288,6 +288,12 @@ func (s *ThreadTurnService) executeSubmission(ctx context.Context, if message.Status != domain.OperatorSteeringPending { return s.refresh(ctx, result) } + if executed.Handoff.Result != nil && executed.Handoff.Result.StopReason == "root_wait" { + // A tool approval keeps the Run running and the exact input prepared. + // Respect the handoff's wait boundary until explicit continuation; + // starting another batch cannot consume that pending approval. + return s.refresh(ctx, result) + } current, err := s.threads.store.GetRun(ctx, result.Submission.Run.ID) if err != nil { return result, apperror.Normalize(err) diff --git a/internal/mcp/tools.go b/internal/mcp/tools.go index 4b8fae5a..7e6e9def 100644 --- a/internal/mcp/tools.go +++ b/internal/mcp/tools.go @@ -6,9 +6,9 @@ import ( "errors" "fmt" "io" + "unicode/utf8" "cyberagent-workbench/internal/domain" - "cyberagent-workbench/internal/idgen" "cyberagent-workbench/internal/toolgateway" ) @@ -44,15 +44,24 @@ func (s *Server) serveToolsCall(ctx context.Context, request Request, output io. } // The client can only supply the closed tool payload; an arbitrary // executable, path, credential, or permission tier has no field to ride on. - call := toolgateway.ToolCall{Name: name, Payload: params.Arguments, - InvocationID: idgen.New("mcp-call"), OperationKey: string(request.ID), + arguments, err := decodeWorkspaceReadArguments(name, params.Arguments) + if err != nil { + _ = s.write(output, Envelope{JSONRPC: "2.0", ID: request.ID, + Error: &RPCError{Code: CodeInvalidParams, Message: err.Error()}}) + return + } + // Workspace reads use the Gateway's Arguments contract. The Gateway owns + // invocation identity and budget charging; a transport request id is not one. + call := toolgateway.ToolCall{Name: name, Arguments: arguments, RunID: s.runID, SessionID: "", WorkspaceID: s.workspaceID, RequestedBy: "mcp:" + s.clientName} - if _, err := s.store.GetRun(ctx, s.runID); err != nil { + run, err := s.store.GetRun(ctx, s.runID) + if err != nil { _ = s.write(output, Envelope{JSONRPC: "2.0", ID: request.ID, Error: &RPCError{Code: CodeInternalError, Message: "Run scope unavailable"}}) return } + call.SessionID = run.SessionID outcome, err := s.tools.Invoke(ctx, call) if err != nil { reason := "invocation_failed" @@ -94,5 +103,24 @@ func (s *Server) serveToolsCall(ctx context.Context, request Request, output io. "tool": params.Name, "status": outcome.Result.Status}) } +func decodeWorkspaceReadArguments(name toolgateway.ToolName, raw json.RawMessage) (map[string]string, error) { + var arguments map[string]string + if err := json.Unmarshal(raw, &arguments); err != nil || arguments == nil { + return nil, errors.New("workspace read arguments must be a JSON object of strings") + } + for key, value := range arguments { + if key != "path" { + return nil, errors.New("workspace read arguments only accept path") + } + if value == "" || utf8.RuneCountInString(value) > 2048 { + return nil, errors.New("workspace read path must contain 1 to 2048 characters") + } + } + if name == toolgateway.ReadFileTool && arguments["path"] == "" { + return nil, errors.New("read_file requires path") + } + return arguments, nil +} + var _ = domain.ValidAgentID var _ = fmt.Sprintf