mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-10 02:16:45 +08:00
* fix(daemon): defer issue status rules to the project workflow (MUL-8145) When the claim carries a project workflow, the runtime brief no longer tells the agent to set in_progress first and in_review on delivery. Those statuses may be outside the workflow (the server rejects them), and entering a handoff step reassigns the issue. The brief now points at the per-turn [Project workflow] block, and the workflow brief carries the entry/exit checks in workflow terms. Briefs without a workflow are byte-identical to before. Co-authored-by: multica-agent <[email protected]> * fix(squad): word the owning leader's status moves by the project workflow (MUL-8145) The squad protocol (Agent Identity, which outranks the runtime brief) still told an owning leader to move the parent to in_progress and then in_review, which a workflow may not list and which can hand the parent to another step's handler. When the claim carries a project workflow, responsibility 6 now keeps the ownership grant but moves the parent only through the [Project workflow] commands. The claim resolves the workflow before the briefing to choose the wording. Guest leaders and issues without a workflow are unchanged. Co-authored-by: multica-agent <[email protected]> * refactor(daemon): list no status keys in the brief for a workflow issue (MUL-8145) With a project workflow, [Project workflow] already lists the issue's statuses and the commands that move it, and the built-in keys may not be accepted there. The status command line now keeps only the command and points at that block; other issues learn valid statuses from the error. Co-authored-by: multica-agent <[email protected]> * refactor(workflow): stop forbidding status moves nothing asks for (MUL-8145) Under a project workflow the current runtime brief no longer states the built-in in_progress / in_review moves, so the lines that forbade them (in [Project workflow], the runtime brief and the squad protocol) only named statuses the agent had no reason to use. Drop them, and the exit check the runtime brief already states. Older daemons still state the built-in rules, so the daemon now advertises workflow-status-rules-v1; without it the claim keeps the override lines in [Project workflow] and the squad protocol. The multica-platform skill notes that [Project workflow] supersedes its built-in status arc. Co-authored-by: multica-agent <[email protected]> * refactor(workflow): shorten the step's moves in the workflow brief (MUL-8145) Replace the handoff preamble with a plain "Your moves:" heading, name who each move hands off to on its own line, and drop the "this workflow has no blocked status" explanations from Can't proceed. Co-authored-by: multica-agent <[email protected]> * refactor(workflow): state each step outcome as the status to set (MUL-8145) The moves block read like raw CLI ('Step done → multica issue status MUL-7 x'). Each outcome now says which status to set and that the server assigns the next handler; the CLI form is given once. Co-authored-by: multica-agent <[email protected]> * Revert "refactor(workflow): state each step outcome as the status to set (MUL-8145)" This reverts commit 2d82d2d342a99afe88cc40c48516b7ea8e860d1e. Co-authored-by: multica-agent <[email protected]> --------- Co-authored-by: multica-eve <[email protected]> Co-authored-by: multica-agent <[email protected]>
654 lines
22 KiB
Go
654 lines
22 KiB
Go
package daemon
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"runtime"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
|
|
"github.com/multica-ai/multica/server/pkg/remotemcp"
|
|
"time"
|
|
|
|
"github.com/multica-ai/multica/server/pkg/protocol"
|
|
)
|
|
|
|
func TestClient_IdentityHeaders_PostJSON(t *testing.T) {
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if got := r.Header.Get("X-Client-Platform"); got != "daemon" {
|
|
t.Errorf("expected X-Client-Platform daemon, got %q", got)
|
|
}
|
|
if got := r.Header.Get("X-Client-Version"); got != "9.9.9" {
|
|
t.Errorf("expected X-Client-Version 9.9.9, got %q", got)
|
|
}
|
|
if got := r.Header.Get("X-Client-OS"); got != normalizeGOOS(runtime.GOOS) {
|
|
t.Errorf("expected X-Client-OS %q, got %q", normalizeGOOS(runtime.GOOS), got)
|
|
}
|
|
if got := r.Header.Get("Authorization"); got != "Bearer tok" {
|
|
t.Errorf("expected Authorization Bearer tok, got %q", got)
|
|
}
|
|
capabilities := make(map[string]bool)
|
|
for _, capability := range strings.Split(r.Header.Get("X-Client-Capabilities"), ",") {
|
|
capabilities[strings.TrimSpace(capability)] = true
|
|
}
|
|
for _, want := range []string{
|
|
protocol.DaemonCapabilitySkillBundlesV1,
|
|
protocol.DaemonCapabilityCoalescedCommentsV1,
|
|
// The worktree gate is decided entirely from this header: if the
|
|
// daemon stops advertising it, every worktree task on this machine
|
|
// is cancelled with an upgrade prompt (MUL-5707). Pin it here so
|
|
// dropping it from the list can never be a silent change.
|
|
protocol.DaemonCapabilityLocalWorktreeV1,
|
|
// Same shape, opposite default: this daemon's brief names the
|
|
// merged multica-platform skill, and advertising that is what
|
|
// stops the server shipping it a redirect stub under the old name
|
|
// (MUL-6986). Dropping it would silently hand every task on this
|
|
// machine a skill it does not need; the failure is extra payload
|
|
// and a stale signpost, neither of which any other test would
|
|
// notice.
|
|
protocol.DaemonCapabilityPlatformSkillV1,
|
|
// Gates whether an automatic retry is handed its parent's workdir
|
|
// (MUL-7034). Dropping it silently sends those retries back to a
|
|
// fresh directory, losing the continuity nothing else would flag.
|
|
protocol.DaemonCapabilityCheckoutKeepsWorkV1,
|
|
// Without it the server never hands this daemon the wakeups that
|
|
// waited for its run; they start runs of their own instead.
|
|
protocol.DaemonCapabilityJoinedWakeupsV1,
|
|
// Without it the server keeps adding the lines that override the
|
|
// built-in in_progress / in_review rules to [Project workflow],
|
|
// rules this daemon's brief no longer states (MUL-8145).
|
|
protocol.DaemonCapabilityWorkflowStatusRulesV1,
|
|
} {
|
|
if !capabilities[want] {
|
|
t.Errorf("X-Client-Capabilities missing %q: %v", want, capabilities)
|
|
}
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(map[string]string{"ok": "1"})
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
c.SetToken("tok")
|
|
c.SetVersion("9.9.9")
|
|
|
|
if err := c.postJSON(context.Background(), "/api/daemon/test", map[string]any{}, nil); err != nil {
|
|
t.Fatalf("postJSON: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestClient_IdentityHeaders_GetJSON(t *testing.T) {
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if got := r.Header.Get("X-Client-Platform"); got != "daemon" {
|
|
t.Errorf("expected X-Client-Platform daemon, got %q", got)
|
|
}
|
|
if got := r.Header.Get("X-Client-Version"); got != "1.2.3" {
|
|
t.Errorf("expected X-Client-Version 1.2.3, got %q", got)
|
|
}
|
|
if got := r.Header.Get("X-Client-OS"); got == "" {
|
|
t.Errorf("expected X-Client-OS to be set")
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Write([]byte(`{}`))
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
c.SetToken("tok")
|
|
c.SetVersion("1.2.3")
|
|
|
|
var out map[string]any
|
|
if err := c.getJSON(context.Background(), "/api/daemon/test", &out); err != nil {
|
|
t.Fatalf("getJSON: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestStartTaskCapabilityNegotiationMixedVersions(t *testing.T) {
|
|
defer noSleepRetry(t)()
|
|
const prefix = `{"supplement_capability":"task-supplement-v1","issue":{"description":"`
|
|
const suffix = `"}}`
|
|
responseAtLimit := prefix + strings.Repeat("x", (1<<20)-len(prefix)-len(suffix)) + suffix
|
|
for _, tc := range []struct {
|
|
name string
|
|
response string
|
|
contentLength string
|
|
negotiated bool
|
|
wantError bool
|
|
retry bool
|
|
}{
|
|
{name: "empty response", response: ""},
|
|
{name: "truncated HTTP body", contentLength: "8", wantError: true, retry: true},
|
|
{name: "truncated response", response: `{"supplement_capability":`, wantError: true},
|
|
{name: "trailing garbage", response: `{"supplement_capability":"task-supplement-v1"}garbage`, wantError: true},
|
|
{name: "HTML response", response: `<html>Proxy error</html>`, wantError: true},
|
|
{name: "response at size limit", response: responseAtLimit, negotiated: true},
|
|
{name: "response exceeds size limit", response: responseAtLimit + " ", wantError: true},
|
|
{name: "old server task response", response: `{"id":"task-1","status":"running"}`, negotiated: false},
|
|
{name: "new server explicit capability", response: `{"supplement_capability":"task-supplement-v1"}`, negotiated: true},
|
|
} {
|
|
for _, mode := range []string{"legacy", "claim-fenced"} {
|
|
t.Run(tc.name+"/"+mode, func(t *testing.T) {
|
|
var calls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
calls.Add(1)
|
|
var body struct {
|
|
Capabilities []string `json:"capabilities"`
|
|
}
|
|
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
|
t.Errorf("decode start request: %v", err)
|
|
}
|
|
if len(body.Capabilities) != 1 || body.Capabilities[0] != protocol.DaemonCapabilityTaskSupplementV1 {
|
|
t.Errorf("capabilities = %#v", body.Capabilities)
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if tc.contentLength != "" {
|
|
w.Header().Set("Content-Length", tc.contentLength)
|
|
}
|
|
_, _ = w.Write([]byte(tc.response))
|
|
}))
|
|
defer srv.Close()
|
|
|
|
task := startTestClaim()
|
|
task.StartClaimSupported = mode == "claim-fenced"
|
|
got, err := NewClient(srv.URL).StartTask(context.Background(), task, protocol.DaemonCapabilityTaskSupplementV1)
|
|
if (err != nil) != tc.wantError {
|
|
t.Errorf("StartTask: %v", err)
|
|
}
|
|
if got != tc.negotiated {
|
|
t.Errorf("negotiated = %v, want %v", got, tc.negotiated)
|
|
}
|
|
wantCalls := 1
|
|
if tc.retry && task.StartClaimSupported {
|
|
wantCalls += len(startTaskRetrySchedule)
|
|
}
|
|
if got := int(calls.Load()); got != wantCalls {
|
|
t.Errorf("requests = %d, want %d", got, wantCalls)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestClient_ResolveRemoteMCPCredentialUsesExplicitDaemonToken(t *testing.T) {
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if got := r.Header.Get("Authorization"); got != "Bearer mdt_task_broker" {
|
|
t.Errorf("Authorization = %q, want short-lived daemon token", got)
|
|
}
|
|
if got := r.URL.Path; got != "/api/daemon/tasks/task-1/remote-mcp/contribution-1/credential" {
|
|
t.Errorf("path = %q", got)
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`{"credential_header":"Authorization","credential":"Bearer upstream"}`))
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
c.SetToken("mul_owner_pat")
|
|
headers, err := c.ResolveRemoteMCPCredential(context.Background(), "mdt_task_broker", "task-1", "contribution-1")
|
|
if err != nil {
|
|
t.Fatalf("ResolveRemoteMCPCredential: %v", err)
|
|
}
|
|
if got := headers.Get("Authorization"); got != "Bearer upstream" {
|
|
t.Fatalf("resolved credential = %q", got)
|
|
}
|
|
if got := c.Token(); got != "mul_owner_pat" {
|
|
t.Fatalf("client PAT was mutated to %q", got)
|
|
}
|
|
}
|
|
|
|
// A Plugin's mcp hook shares this resolver and this broker with a workspace's
|
|
// own Remote MCP connections, but its credential lives in the Plugin's secret
|
|
// storage and a different route serves it. The contribution id is all the
|
|
// broker hands back at dial time, so the id carries the marker — and a
|
|
// connection without it must keep going to the original route.
|
|
func TestClient_ResolveRemoteMCPCredentialRoutesPluginContributions(t *testing.T) {
|
|
var seen []string
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
seen = append(seen, r.URL.Path)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`{"credential_header":"Authorization","credential":"Bearer upstream"}`))
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
for _, contribution := range []string{"contribution-1", remotemcp.PluginContributionPrefix + "install-1:toolbox"} {
|
|
if _, err := c.ResolveRemoteMCPCredential(context.Background(), "mdt_task_broker", "task-1", contribution); err != nil {
|
|
t.Fatalf("resolve %q: %v", contribution, err)
|
|
}
|
|
}
|
|
|
|
want := []string{
|
|
"/api/daemon/tasks/task-1/remote-mcp/contribution-1/credential",
|
|
"/api/daemon/tasks/task-1/plugin-mcp/plugin:install-1:toolbox/credential",
|
|
}
|
|
for i, path := range want {
|
|
if seen[i] != path {
|
|
t.Fatalf("request %d went to %q, want %q", i, seen[i], path)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestClient_VersionOmittedWhenUnset(t *testing.T) {
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if got := r.Header.Get("X-Client-Platform"); got != "daemon" {
|
|
t.Errorf("expected X-Client-Platform daemon, got %q", got)
|
|
}
|
|
// SetVersion not called → header must be omitted (not "").
|
|
if vals := r.Header.Values("X-Client-Version"); len(vals) != 0 {
|
|
t.Errorf("expected X-Client-Version absent, got %v", vals)
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
if err := c.postJSON(context.Background(), "/api/daemon/test", nil, nil); err != nil {
|
|
t.Fatalf("postJSON: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestClient_ListWorkspacesUsesDaemonEndpointAndETag(t *testing.T) {
|
|
var calls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != "/api/daemon/workspaces" {
|
|
t.Errorf("path = %q, want /api/daemon/workspaces", r.URL.Path)
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
call := calls.Add(1)
|
|
if call == 1 {
|
|
if got := r.Header.Get("If-None-Match"); got != "" {
|
|
t.Errorf("first If-None-Match = %q, want empty", got)
|
|
}
|
|
w.Header().Set("ETag", `W/"workspace-v1"`)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`[{"id":"ws-1","name":"One"}]`))
|
|
return
|
|
}
|
|
if got := r.Header.Get("If-None-Match"); got != `W/"workspace-v1"` {
|
|
t.Errorf("If-None-Match = %q, want cached ETag", got)
|
|
}
|
|
w.WriteHeader(http.StatusNotModified)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
first, err := c.ListWorkspaces(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("first ListWorkspaces: %v", err)
|
|
}
|
|
second, err := c.ListWorkspaces(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("second ListWorkspaces: %v", err)
|
|
}
|
|
if len(first) != 1 || len(second) != 1 || second[0] != first[0] {
|
|
t.Fatalf("cached workspaces mismatch: first=%+v second=%+v", first, second)
|
|
}
|
|
}
|
|
|
|
func TestClient_ListWorkspacesFallsBackToLegacyEndpointOnce(t *testing.T) {
|
|
var daemonCalls atomic.Int32
|
|
var legacyCalls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
switch r.URL.Path {
|
|
case "/api/daemon/workspaces":
|
|
daemonCalls.Add(1)
|
|
http.NotFound(w, r)
|
|
case "/api/workspaces":
|
|
legacyCalls.Add(1)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`[{"id":"ws-legacy","name":"Legacy"}]`))
|
|
default:
|
|
http.NotFound(w, r)
|
|
}
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
for i := 0; i < 2; i++ {
|
|
workspaces, err := c.ListWorkspaces(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("ListWorkspaces call %d: %v", i+1, err)
|
|
}
|
|
if len(workspaces) != 1 || workspaces[0].ID != "ws-legacy" {
|
|
t.Fatalf("workspaces = %+v, want legacy response", workspaces)
|
|
}
|
|
}
|
|
if got := daemonCalls.Load(); got != 1 {
|
|
t.Fatalf("daemon endpoint calls = %d, want 1", got)
|
|
}
|
|
if got := legacyCalls.Load(); got != 2 {
|
|
t.Fatalf("legacy endpoint calls = %d, want 2", got)
|
|
}
|
|
}
|
|
|
|
// noSleepRetry replaces retrySleep with an immediate no-op so tests don't
|
|
// actually wait the 4s/8s/16s/... backoffs. Returns a restore func.
|
|
func noSleepRetry(t *testing.T) func() {
|
|
t.Helper()
|
|
prev := retrySleep
|
|
retrySleep = func(ctx context.Context, _ time.Duration) error {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
return func() { retrySleep = prev }
|
|
}
|
|
|
|
func TestIsTransientError(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
err error
|
|
want bool
|
|
}{
|
|
{"nil is not transient", nil, false},
|
|
{"5xx is transient", &requestError{StatusCode: http.StatusBadGateway}, true},
|
|
{"503 is transient", &requestError{StatusCode: http.StatusServiceUnavailable}, true},
|
|
{"408 is transient", &requestError{StatusCode: http.StatusRequestTimeout}, true},
|
|
{"429 is transient", &requestError{StatusCode: http.StatusTooManyRequests}, true},
|
|
{"400 is permanent", &requestError{StatusCode: http.StatusBadRequest}, false},
|
|
{"401 is permanent", &requestError{StatusCode: http.StatusUnauthorized}, false},
|
|
{"404 is permanent", &requestError{StatusCode: http.StatusNotFound}, false},
|
|
{"transport-level error is transient", errors.New("connection reset by peer"), true},
|
|
}
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
if got := isTransientError(tc.err); got != tc.want {
|
|
t.Fatalf("isTransientError(%v) = %v, want %v", tc.err, got, tc.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestIsIssueGCBatchUnsupported(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
err error
|
|
want bool
|
|
}{
|
|
{
|
|
name: "old server unmatched route",
|
|
err: &requestError{StatusCode: http.StatusNotFound, Body: "404 page not found"},
|
|
want: true,
|
|
},
|
|
{
|
|
name: "workspace access denied",
|
|
err: &requestError{StatusCode: http.StatusNotFound, Body: `{"error":"not found"}`},
|
|
want: false,
|
|
},
|
|
{
|
|
name: "transient server error",
|
|
err: &requestError{StatusCode: http.StatusInternalServerError, Body: "failure"},
|
|
want: false,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
if got := isIssueGCBatchUnsupported(tt.err); got != tt.want {
|
|
t.Fatalf("isIssueGCBatchUnsupported() = %v, want %v", got, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPostJSONWithRetry_TransientThenSuccess(t *testing.T) {
|
|
defer noSleepRetry(t)()
|
|
|
|
var calls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
n := calls.Add(1)
|
|
if n < 3 {
|
|
w.WriteHeader(http.StatusBadGateway)
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
schedule := []time.Duration{time.Nanosecond, time.Nanosecond, time.Nanosecond}
|
|
if err := c.postJSONWithRetry(context.Background(), "/x", map[string]any{}, nil, schedule); err != nil {
|
|
t.Fatalf("postJSONWithRetry: %v", err)
|
|
}
|
|
if got := calls.Load(); got != 3 {
|
|
t.Fatalf("expected 3 attempts (2 transient + 1 success), got %d", got)
|
|
}
|
|
}
|
|
|
|
// TestFailTask_RetriesOnTransient5xxThenSucceeds pins the callback half of
|
|
// MUL-5305 Must-fix 1: FailTask's terminal transaction is now the sole
|
|
// persistence point for the withheld session and continuity-gap flag, so if the
|
|
// server returns a transient 5xx (the terminal tx rolled back), the daemon MUST
|
|
// retry until it lands — a 400 would make it bail immediately
|
|
// (TestPostJSONWithRetry_PermanentBailsImmediately) and drop the gap forever.
|
|
func TestFailTask_RetriesOnTransient5xxThenSucceeds(t *testing.T) {
|
|
defer noSleepRetry(t)()
|
|
|
|
var calls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
if calls.Add(1) < 3 {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
if err := c.FailTask(context.Background(), "task-1", "boom", "", "", "", "timeout", true, "", ""); err != nil {
|
|
t.Fatalf("FailTask: %v", err)
|
|
}
|
|
if got := calls.Load(); got != 3 {
|
|
t.Fatalf("expected 3 attempts (2 transient 5xx + 1 success), got %d", got)
|
|
}
|
|
}
|
|
|
|
func TestPostJSONWithRetry_TransientExhausts(t *testing.T) {
|
|
defer noSleepRetry(t)()
|
|
|
|
var calls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
calls.Add(1)
|
|
w.WriteHeader(http.StatusBadGateway)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
schedule := []time.Duration{time.Nanosecond, time.Nanosecond}
|
|
err := c.postJSONWithRetry(context.Background(), "/x", map[string]any{}, nil, schedule)
|
|
if err == nil {
|
|
t.Fatal("expected error after schedule exhausted, got nil")
|
|
}
|
|
if !isTransientError(err) {
|
|
t.Fatalf("expected transient error, got %v", err)
|
|
}
|
|
if got := calls.Load(); got != int32(len(schedule)+1) {
|
|
t.Fatalf("expected %d attempts (initial + %d retries), got %d", len(schedule)+1, len(schedule), got)
|
|
}
|
|
}
|
|
|
|
func TestPostJSONWithRetry_PermanentBailsImmediately(t *testing.T) {
|
|
defer noSleepRetry(t)()
|
|
|
|
var calls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
calls.Add(1)
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
schedule := []time.Duration{time.Nanosecond, time.Nanosecond, time.Nanosecond}
|
|
err := c.postJSONWithRetry(context.Background(), "/x", map[string]any{}, nil, schedule)
|
|
if err == nil {
|
|
t.Fatal("expected error, got nil")
|
|
}
|
|
if got := calls.Load(); got != 1 {
|
|
t.Fatalf("expected exactly 1 attempt on permanent error, got %d", got)
|
|
}
|
|
}
|
|
|
|
func TestPostJSONWithRetry_CtxCancelStopsRetries(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
// Use the real sleeper here so we can observe a cancel preempting it.
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
var calls atomic.Int32
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
calls.Add(1)
|
|
w.WriteHeader(http.StatusBadGateway)
|
|
w.(http.Flusher).Flush()
|
|
// Cancel only once the first attempt has been answered: it lands while
|
|
// the client finishes that response or in the 1s retry sleep after it,
|
|
// never before the first attempt, and no second attempt can start.
|
|
cancel()
|
|
}))
|
|
defer srv.Close()
|
|
|
|
c := NewClient(srv.URL)
|
|
schedule := []time.Duration{time.Second, time.Second, time.Second}
|
|
start := time.Now()
|
|
err := c.postJSONWithRetry(ctx, "/x", map[string]any{}, nil, schedule)
|
|
elapsed := time.Since(start)
|
|
if err == nil {
|
|
t.Fatal("expected error after ctx cancel, got nil")
|
|
}
|
|
if elapsed > 750*time.Millisecond {
|
|
t.Fatalf("expected ctx cancel to short-circuit retry, took %s", elapsed)
|
|
}
|
|
if got := calls.Load(); got != 1 {
|
|
t.Fatalf("expected exactly 1 attempt before cancel, got %d", got)
|
|
}
|
|
}
|
|
|
|
func TestNormalizeGOOS(t *testing.T) {
|
|
cases := map[string]string{
|
|
"darwin": "macos",
|
|
"windows": "windows",
|
|
"linux": "linux",
|
|
"freebsd": "freebsd",
|
|
}
|
|
for in, want := range cases {
|
|
if got := normalizeGOOS(in); got != want {
|
|
t.Errorf("normalizeGOOS(%q) = %q, want %q", in, got, want)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestTerminalReportsCarryRetiredSessionID pins the daemon half of the
|
|
// retire-session contract (GH #6066). Before it, a terminal report could only
|
|
// say "here is a session" or say nothing — and saying nothing was how a
|
|
// recovered turn silently left the poisoned id selectable. The completed path
|
|
// matters most: that is exactly the case where a fresh-session retry SUCCEEDED
|
|
// and the abandoned transcript would otherwise survive on an older row.
|
|
func TestTerminalReportsCarryRetiredSessionID(t *testing.T) {
|
|
for _, tc := range []struct {
|
|
name string
|
|
endpoint string
|
|
call func(*Client) error
|
|
}{
|
|
{
|
|
name: "complete",
|
|
endpoint: "/api/daemon/tasks/task-1/complete",
|
|
call: func(c *Client) error {
|
|
return c.CompleteTask(context.Background(), "task-1", "done", "", "", "/tmp/wd", false, "POISONED-S", "")
|
|
},
|
|
},
|
|
{
|
|
name: "fail",
|
|
endpoint: "/api/daemon/tasks/task-1/fail",
|
|
call: func(c *Client) error {
|
|
return c.FailTask(context.Background(), "task-1", "boom", "", "/tmp/wd", "", "api_invalid_request", false, "POISONED-S", "")
|
|
},
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
var body map[string]any
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != tc.endpoint {
|
|
t.Errorf("unexpected path %q", r.URL.Path)
|
|
}
|
|
_ = json.NewDecoder(r.Body).Decode(&body)
|
|
w.WriteHeader(http.StatusOK)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
if err := tc.call(NewClient(srv.URL)); err != nil {
|
|
t.Fatalf("terminal report: %v", err)
|
|
}
|
|
if got, _ := body["retired_session_id"].(string); got != "POISONED-S" {
|
|
t.Fatalf("retired_session_id = %v, want POISONED-S (body: %v)", body["retired_session_id"], body)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestTerminalReportsOmitEmptyRetiredSessionID keeps the common case off the
|
|
// wire: nearly every run retires nothing, and an empty field would be
|
|
// indistinguishable from "retire the empty session".
|
|
func TestTerminalReportsOmitEmptyRetiredSessionID(t *testing.T) {
|
|
var body map[string]any
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
_ = json.NewDecoder(r.Body).Decode(&body)
|
|
w.WriteHeader(http.StatusOK)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
if err := NewClient(srv.URL).CompleteTask(context.Background(), "task-1", "done", "", "sess-1", "/tmp/wd", false, "", ""); err != nil {
|
|
t.Fatalf("CompleteTask: %v", err)
|
|
}
|
|
if _, present := body["retired_session_id"]; present {
|
|
t.Fatalf("retired_session_id must be omitted when nothing was retired, got %v", body)
|
|
}
|
|
}
|
|
|
|
func TestTerminalReportsCarryDurableWorkDir(t *testing.T) {
|
|
const durableWorkDir = "/Users/dev/project"
|
|
for _, tc := range []struct {
|
|
name string
|
|
call func(*Client) error
|
|
}{
|
|
{
|
|
name: "complete",
|
|
call: func(c *Client) error {
|
|
return c.CompleteTask(context.Background(), "task-1", "done", "", "", "/tmp/wd", false, "", durableWorkDir)
|
|
},
|
|
},
|
|
{
|
|
name: "fail",
|
|
call: func(c *Client) error {
|
|
return c.FailTask(context.Background(), "task-1", "boom", "", "/tmp/wd", "", "agent_error", false, "", durableWorkDir)
|
|
},
|
|
},
|
|
{
|
|
name: "cancel ack",
|
|
call: func(c *Client) error {
|
|
return c.AckTaskCancelled(context.Background(), "task-1", TaskCancelAck{DurableWorkDir: durableWorkDir})
|
|
},
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
var body map[string]any
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
_ = json.NewDecoder(r.Body).Decode(&body)
|
|
w.WriteHeader(http.StatusOK)
|
|
}))
|
|
defer srv.Close()
|
|
|
|
if err := tc.call(NewClient(srv.URL)); err != nil {
|
|
t.Fatalf("terminal report: %v", err)
|
|
}
|
|
if got := body["durable_work_dir"]; got != durableWorkDir {
|
|
t.Fatalf("durable_work_dir = %v, want %q (body: %v)", got, durableWorkDir, body)
|
|
}
|
|
})
|
|
}
|
|
}
|