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") }