From be08938f1fa635c47713a1cdbebe51cd94c4c432 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 30 Jul 2026 23:39:50 +0400 Subject: [PATCH] Thread context through Provider.Execute and stop reporting read failures as unknown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Finding 6 of REVIEW-2026-07-30.md. runWithTimeout could not cancel anything, because Provider.Execute took no context: the goroutine ran on to the HTTP client's 60s timeout, outliving the 30s capability timeout. Provider.Execute now takes a context carrying that timeout, and the workspace provider issues its tool call with http.NewRequestWithContext, so a timed-out execution actually tears the request down. A read-only capability whose provider call times out now resolves to failed rather than unknown. Spec §4.3 reserves unknown for executions whose side effect may or may not have landed, and never retries them — which made read failures both unretryable and indistinguishable from genuinely ambiguous mutations, for calls that by definition have no side effect. Mutating capabilities still resolve to unknown. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Uea55zaiWuEByEDC4UBSdd --- internal/execution/engine.go | 45 +++++++++----- internal/execution/engine_test.go | 78 +++++++++++++++++++++++-- internal/provider/registry.go | 7 ++- internal/provider/workspace_mcp.go | 26 +++++---- internal/provider/workspace_mcp_test.go | 5 +- 5 files changed, 129 insertions(+), 32 deletions(-) diff --git a/internal/execution/engine.go b/internal/execution/engine.go index ea60394..4d4652a 100644 --- a/internal/execution/engine.go +++ b/internal/execution/engine.go @@ -218,14 +218,23 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { switch { case timedOut: - // Outcome is never "failed" on timeout, and never retried automatically - // per ECOSYSTEM-SPEC.md §4.3 — the side effect may or may not have landed. - exec.Status = domain.ExecutionUnknown + // ECOSYSTEM-SPEC.md §4.3 reserves "unknown" for executions whose side + // effect may or may not have landed — such an execution is never + // retried automatically. That reasoning only applies to mutations. A + // read_only capability has no side effect by definition, so a timed-out + // read is unambiguously a failure and is safe to retry; reporting it as + // "unknown" would both mislead and strand it. exec.Error = "execution timed out after " + timeout.String() exec.Result = map[string]any{"error": exec.Error} + outcome := "unknown" + exec.Status = domain.ExecutionUnknown + if capability.ReadOnly { + outcome = "failed" + exec.Status = domain.ExecutionFailed + } e.emitEventCorrelated(domain.EventExecutionFailed, exec.ID, exec.CorrelationID, exec.CausationID, map[string]any{ "capability_id": req.CapabilityID, - "outcome": "unknown", + "outcome": outcome, "reason": "timeout", }) case execErr != nil: @@ -251,25 +260,35 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { return &ExecuteResult{Execution: exec}, nil } -// runWithTimeout executes the provider call on a goroutine and returns -// (nil, nil, true) if it doesn't finish within timeout. The goroutine is not -// cancelled — providers don't accept a context today — so a slow executor -// keeps running in the background; its eventual result is discarded, never -// retried, and the caller has already moved on with outcome=unknown. +// runWithTimeout executes the provider call under a context carrying the +// capability's timeout, and returns (nil, nil, true) if it doesn't finish in +// time. Unlike the previous implementation, the context is handed to the +// provider, so a timeout genuinely cancels the underlying work (an in-flight +// HTTP request is torn down) rather than abandoning a goroutine that runs on +// to whatever longer timeout the provider's own client happens to use. +// +// The goroutine still exists — a provider that ignores its context can only +// be waited on, not killed — but the channel is buffered, so it can always +// deliver and exit; nothing is leaked permanently. +// +// This timeout is deliberately separate from, and downstream of, the bounded +// Nexus lookup in validateTarget: target validation has already completed by +// the time we get here, so the two never nest. func (e *Engine) runWithTimeout(prov provider.Provider, capability *domain.Capability, req *domain.ExecuteRequest, timeout time.Duration) (map[string]any, error, bool) { type outcome struct { result map[string]any err error } ch := make(chan outcome, 1) - go func() { - r, err := prov.Execute(capability, req) - ch <- outcome{result: r, err: err} - }() ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() + go func() { + r, err := prov.Execute(ctx, capability, req) + ch <- outcome{result: r, err: err} + }() + select { case o := <-ch: return o.result, o.err, false diff --git a/internal/execution/engine_test.go b/internal/execution/engine_test.go index da798c2..4cdc6db 100644 --- a/internal/execution/engine_test.go +++ b/internal/execution/engine_test.go @@ -1,6 +1,7 @@ package execution_test import ( + "context" "errors" "path/filepath" "testing" @@ -16,13 +17,29 @@ type fakeProvider struct { name string delay time.Duration err error + + // returned closes once Execute has actually returned, letting a test + // prove the provider call was cancelled rather than left running. + returned chan struct{} + // ctxErr records the context error observed by Execute, if any. + ctxErr error } func (p *fakeProvider) Name() string { return p.name } -func (p *fakeProvider) Execute(capability *domain.Capability, req *domain.ExecuteRequest) (map[string]any, error) { +func (p *fakeProvider) Execute(ctx context.Context, capability *domain.Capability, req *domain.ExecuteRequest) (map[string]any, error) { + if p.returned != nil { + defer close(p.returned) + } if p.delay > 0 { - time.Sleep(p.delay) + // A well-behaved provider aborts as soon as its context is done — + // this stands in for tearing down an in-flight HTTP request. + select { + case <-time.After(p.delay): + case <-ctx.Done(): + p.ctxErr = ctx.Err() + return nil, ctx.Err() + } } if p.err != nil { return nil, p.err @@ -286,8 +303,57 @@ func TestExecute_TimeoutYieldsUnknownNotFailed(t *testing.T) { if res.Execution.Status != domain.ExecutionUnknown { t.Fatalf("expected status unknown on timeout, got %s", res.Execution.Status) } - - // Let the background provider goroutine finish so t.Cleanup can close - // the store without a dangling write racing it. - time.Sleep(1600 * time.Millisecond) +} + +// A read_only capability has no side effect, so a timeout is an unambiguous +// failure: retryable, and distinguishable from a mutation that may or may not +// have landed. ECOSYSTEM-SPEC.md §4.3's "unknown" outcome must not be used +// for reads. +func TestExecute_ReadOnlyTimeoutYieldsFailedNotUnknown(t *testing.T) { + prov := &fakeProvider{name: "slow", delay: 1500 * time.Millisecond} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.TimeoutSeconds = 1 + c.Provider = "slow" + c.Risk = "read" + c.ReadOnly = true + }) + + res, err := eng.Execute(&domain.ExecuteRequest{CapabilityID: cap.ID, TargetEntityID: "ent_x"}) + if err != nil { + t.Fatalf("execute: %v", err) + } + if res.Execution.Status != domain.ExecutionFailed { + t.Fatalf("expected status failed on read-only timeout, got %s", res.Execution.Status) + } + if res.Execution.Error == "" { + t.Fatal("expected a timeout error message on the execution") + } +} + +// The capability timeout must actually cancel the provider call. Before the +// context was threaded through Provider.Execute, the goroutine ran on to the +// provider's own (much longer) client timeout; here it must observe +// cancellation and return promptly after the 1s capability timeout, well +// before its nominal 30s delay. +func TestExecute_TimeoutCancelsProviderCall(t *testing.T) { + prov := &fakeProvider{name: "slow", delay: 30 * time.Second, returned: make(chan struct{})} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.TimeoutSeconds = 1 + c.Provider = "slow" + }) + + if _, err := eng.Execute(&domain.ExecuteRequest{CapabilityID: cap.ID, TargetEntityID: "ent_x"}); err != nil { + t.Fatalf("execute: %v", err) + } + + select { + case <-prov.returned: + case <-time.After(5 * time.Second): + t.Fatal("provider call was not cancelled: it outlived the capability timeout") + } + if !errors.Is(prov.ctxErr, context.DeadlineExceeded) { + t.Fatalf("expected provider to observe context.DeadlineExceeded, got %v", prov.ctxErr) + } } diff --git a/internal/provider/registry.go b/internal/provider/registry.go index 34eb8e6..e24882b 100644 --- a/internal/provider/registry.go +++ b/internal/provider/registry.go @@ -1,6 +1,7 @@ package provider import ( + "context" "fmt" "sync" @@ -9,7 +10,11 @@ import ( type Provider interface { Name() string - Execute(capability *domain.Capability, req *domain.ExecuteRequest) (map[string]any, error) + // Execute performs the capability's side effect. The context carries the + // capability's timeout: implementations MUST propagate it into every + // blocking call they make so that a timed-out execution is genuinely + // cancelled rather than abandoned to run on in the background. + Execute(ctx context.Context, capability *domain.Capability, req *domain.ExecuteRequest) (map[string]any, error) } type Registry struct { diff --git a/internal/provider/workspace_mcp.go b/internal/provider/workspace_mcp.go index 54053c8..c45e502 100644 --- a/internal/provider/workspace_mcp.go +++ b/internal/provider/workspace_mcp.go @@ -2,6 +2,7 @@ package provider import ( "bytes" + "context" "encoding/json" "fmt" "io" @@ -109,13 +110,7 @@ func (p *WorkspaceMCPProvider) DiscoverTools() ([]WorkspaceTool, error) { return result.Tools, nil } -func (p *WorkspaceMCPProvider) DiscoveredTools() []WorkspaceTool { - p.mu.RLock() - defer p.mu.RUnlock() - return p.tools -} - -func (p *WorkspaceMCPProvider) Execute(capability *domain.Capability, req *domain.ExecuteRequest) (map[string]any, error) { +func (p *WorkspaceMCPProvider) Execute(ctx context.Context, capability *domain.Capability, req *domain.ExecuteRequest) (map[string]any, error) { toolName := p.capabilityToTool(capability.Name) if toolName == "" { return nil, fmt.Errorf("no workspace tool mapped for capability %q", capability.Name) @@ -146,11 +141,22 @@ func (p *WorkspaceMCPProvider) Execute(capability *domain.Capability, req *domai } body, _ := json.Marshal(args) - resp, err := p.httpClient.Post( + // The caller's context carries the capability timeout. Building the + // request with it means a cancelled execution tears down the in-flight + // HTTP call instead of leaving it to run to the client's own (much + // longer) timeout. + httpReq, err := http.NewRequestWithContext( + ctx, + http.MethodPost, fmt.Sprintf("%s/api/tool/%s", p.baseURL, toolName), - "application/json", bytes.NewReader(body), ) + if err != nil { + return nil, fmt.Errorf("build workspace tool request: %w", err) + } + httpReq.Header.Set("Content-Type", "application/json") + + resp, err := p.httpClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("call workspace tool %q: %w", toolName, err) } @@ -163,7 +169,7 @@ func (p *WorkspaceMCPProvider) Execute(capability *domain.Capability, req *domai if resp.StatusCode >= 400 { return map[string]any{ - "error": string(respBody), + "error": string(respBody), "http_status": resp.StatusCode, }, fmt.Errorf("workspace tool %q returned %d: %s", toolName, resp.StatusCode, string(respBody)) } diff --git a/internal/provider/workspace_mcp_test.go b/internal/provider/workspace_mcp_test.go index 698550c..1295fee 100644 --- a/internal/provider/workspace_mcp_test.go +++ b/internal/provider/workspace_mcp_test.go @@ -1,6 +1,7 @@ package provider import ( + "context" "encoding/json" "net/http" "net/http/httptest" @@ -29,7 +30,7 @@ func TestExecute_SucceedsOnPlainResult(t *testing.T) { }) cap := &domain.Capability{Name: "ws.do_thing"} - result, err := p.Execute(cap, &domain.ExecuteRequest{}) + result, err := p.Execute(context.Background(), cap, &domain.ExecuteRequest{}) if err != nil { t.Fatalf("expected no error, got %v", err) } @@ -52,7 +53,7 @@ func TestExecute_ErrorEnvelopeReportedAsFailure(t *testing.T) { }) cap := &domain.Capability{Name: "ws.do_thing"} - _, err := p.Execute(cap, &domain.ExecuteRequest{}) + _, err := p.Execute(context.Background(), cap, &domain.ExecuteRequest{}) if err == nil { t.Fatal("expected error for ERROR-coded warning envelope, got nil") }