From 50b44be9802bb8c888f3d6be0c0c8578f2beaa34 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Mon, 7 Sep 2026 05:44:23 +0800 Subject: [PATCH 1/5] fix(dispatch): persist and reconcile Hub task admission outcomes Make pending Hub identity durable before executor admission, retain immutable accepted/rejected outcomes, authorize stored replay scopes and distinguish live contention from uncertain recovery. Retry only proven pre-execution rejections and preserve accepted work without another start. Refs #2349. Co-authored-by: Codex --- .../internal/api/handlers_run_delivery.go | 13 +- edge-server/internal/api/handlers_runs.go | 37 +- .../internal/api/hub_task_replay_test.go | 177 +++++++ edge-server/internal/errcode/codes.go | 8 +- .../internal/lifecycle/mock_executor.go | 6 + .../lifecycle/thread_transcript_test.go | 6 + edge-server/internal/runcontrol/admission.go | 143 ++++++ .../internal/runcontrol/admission_test.go | 143 ++++++ edge-server/internal/runcontrol/runcontrol.go | 88 ++-- .../internal/store/admission_cleanup_test.go | 32 ++ edge-server/internal/store/file_store.go | 22 + edge-server/internal/store/sqlite_store.go | 10 + edge-server/internal/store/store_domain.go | 5 +- .../internal/store/store_interfaces.go | 2 + .../internal/store/store_query_plan.go | 4 +- .../internal/store/store_run_admission.go | 102 ++++ .../store/store_run_admission_test.go | 462 ++++++++++++++++++ edge-server/internal/store/store_types.go | 2 + 18 files changed, 1184 insertions(+), 78 deletions(-) create mode 100644 edge-server/internal/api/hub_task_replay_test.go create mode 100644 edge-server/internal/runcontrol/admission.go create mode 100644 edge-server/internal/runcontrol/admission_test.go create mode 100644 edge-server/internal/store/admission_cleanup_test.go create mode 100644 edge-server/internal/store/store_run_admission.go create mode 100644 edge-server/internal/store/store_run_admission_test.go diff --git a/edge-server/internal/api/handlers_run_delivery.go b/edge-server/internal/api/handlers_run_delivery.go index 40470f7af..9a0c538ca 100644 --- a/edge-server/internal/api/handlers_run_delivery.go +++ b/edge-server/internal/api/handlers_run_delivery.go @@ -39,10 +39,7 @@ func (h *Handler) beginRunDelivery(w http.ResponseWriter, r *http.Request, req r // HTTP and Desktop can use different local thread representations for // the same Hub task. Authorize the actual stored scope too: a capability // for the incoming representation must not expose a different resource. - replayRequest := req - replayRequest.ProjectID = run.ProjectID - replayRequest.ThreadID = run.ThreadID - if err := h.validateCapabilityRequest(r, &replayRequest); err != nil { + if err := h.validateRunReplay(r, req, run); err != nil { errcode.Write(w, err) return nil, true } @@ -57,3 +54,11 @@ func (h *Handler) beginRunDelivery(w http.ResponseWriter, r *http.Request, req r } return nil, true } + +// Cached and durable Hub-task receipts must authorize the actual stored run, +// not only the incoming transport's representation of its scope. +func (h *Handler) validateRunReplay(r *http.Request, req runRequest, run store.Run) *errcode.Error { + req.ProjectID = run.ProjectID + req.ThreadID = run.ThreadID + return h.validateCapabilityRequest(r, &req) +} diff --git a/edge-server/internal/api/handlers_runs.go b/edge-server/internal/api/handlers_runs.go index 9ecb13973..266679d8d 100644 --- a/edge-server/internal/api/handlers_runs.go +++ b/edge-server/internal/api/handlers_runs.go @@ -374,17 +374,22 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { // active-run guarding, run record creation, run.queued publication, and // the executor start/failure state machine — REST and MCP share it. REST // contributes the timeline policy and the adapter context builder. + var replayRunID string run, err := runcontrol.Create(repository, h.Executor, h.Bus, runcontrol.CreateParams{ - ProjectID: req.ProjectID, - ThreadID: req.ThreadID, - Prompt: req.Prompt, - AgentID: req.AgentID, - Model: req.Model, - PermissionMode: req.PermissionMode, - SessionID: req.SessionID, - ContinueLast: req.Continue, - WorkDir: req.WorkDir, - HubTaskID: req.HubTaskID, + ProjectID: req.ProjectID, + ThreadID: req.ThreadID, + Prompt: req.Prompt, + AgentID: req.AgentID, + Model: req.Model, + PermissionMode: req.PermissionMode, + SessionID: req.SessionID, + ContinueLast: req.Continue, + WorkDir: req.WorkDir, + HubTaskID: req.HubTaskID, + AuthorizeReplay: func(existing store.Run) *errcode.Error { + replayRunID = existing.ID + return h.validateRunReplay(r, req, existing) + }, WorkspaceAllowlist: h.WorkspaceAllowlist, AgentExists: func(agentID string) bool { if h.AdapterRegistry == nil { @@ -404,6 +409,9 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { }) if err != nil { if e, ok := err.(*errcode.Error); ok { + if errors.Is(err, errcode.ErrDeliveryBusy) || errors.Is(err, errcode.ErrAdmissionPersistFailed) || errors.Is(err, errcode.ErrTooManyConcurrentRuns) { + w.Header().Set("Retry-After", "1") + } // Enrich the active-run conflict with the conflicting run, // preserving the historical response body shape. if errors.Is(err, errcode.ErrActiveRunExists) { @@ -426,7 +434,14 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { errcode.Write(w, errcode.ErrInternal.WithMessage("run admission receipt could not be committed")) return } - writeSuccess(w, http.StatusAccepted, acceptedResponse(runToResponse(run))) + data := runToResponse(run) + if replayRunID == run.ID { + data["deduplicated"] = true + if req.DeliveryID != "" { + data["deliveryId"] = req.DeliveryID + } + } + writeSuccess(w, http.StatusAccepted, acceptedResponse(data)) } // --------------------------------------------------------------------------- diff --git a/edge-server/internal/api/hub_task_replay_test.go b/edge-server/internal/api/hub_task_replay_test.go new file mode 100644 index 000000000..4f18fa66b --- /dev/null +++ b/edge-server/internal/api/hub_task_replay_test.go @@ -0,0 +1,177 @@ +package api + +import ( + "encoding/json" + "errors" + "github.com/agenthub/edge-server/internal/store" + "net/http" + "strings" + "testing" + "time" + + "github.com/agenthub/edge-server/internal/deliverydedup" + "github.com/agenthub/edge-server/internal/lifecycle" +) + +func TestHubTaskReplay_CacheMissStillAuthorizesOriginalScope(t *testing.T) { + const signingKey = "fixture-capability-signing-key-32-bytes" + executor := &admissionExecutor{} + server, h := newDeliveryTestServer(t, executor, func(h *Handler) { + h.HubJWTSecret = signingKey + h.EdgeDeviceID = "fixture-device" + }) + defer server.Close() + repository := ensureStore(h) + if _, err := repository.CreateProject("proj_other", "Other", ""); err != nil { + t.Fatal(err) + } + if _, err := repository.CreateThread("thread_other", "proj_other", "Other", "", "", ""); err != nil { + t.Fatal(err) + } + post := func(projectID, threadID, deliveryID string) postResult { + body := admissionRunBody(h.WorkspaceAllowlist[0], deliveryID, "task-persisted", map[string]any{"projectId": projectID, "threadId": threadID}) + request, err := http.NewRequest(http.MethodPost, server.URL+"/v1/runs", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-AgentHub-Capability-Token", signCapability(signingKey, "user-fixture", "fixture-device", projectID, "run-start", time.Hour)) + return doReq(t, request) + } + first := post("proj_local", "thread_local", "delivery-first") + if first.status != http.StatusAccepted { + t.Fatalf("first admission: %d %#v", first.status, first.body) + } + // Simulate a fresh process-local cache while preserving the repository. + h.DeliveryDedup = deliverydedup.New(deliverydedup.DefaultCapacity, deliverydedup.DefaultTTL) + replay := post("proj_other", "thread_other", "delivery-after-restart") + if replay.status != http.StatusForbidden || errCode(replay.body) != "capability_token_invalid" { + t.Fatalf("cold Hub-task lookup bypassed actual-run authorization: %d %#v", replay.status, replay.body) + } + if executor.StartCount() != 1 { + t.Fatalf("unauthorized replay started work: %d", executor.StartCount()) + } +} + +func TestHubTaskReplay_RejectedExecutorAdmissionIsNotSuccess(t *testing.T) { + executor := &admissionExecutor{failNext: true, err: errors.New("fixture admission rejected")} + server, h := newDeliveryTestServer(t, executor, nil) + defer server.Close() + body := admissionRunBody(h.WorkspaceAllowlist[0], "delivery-rejected", "task-rejected", nil) + first := postRunsRaw(t, server.URL, body) + if first.status == http.StatusAccepted { + t.Fatalf("expected initial rejection: %#v", first.body) + } + replay := postRunsRaw(t, server.URL, body) + if replay.status != http.StatusInternalServerError || errCode(replay.body) != "executor_start_failed" || executor.StartCount() != 1 { + t.Fatalf("rejected Hub task must retain rejection without restarting: %d %#v starts=%d", replay.status, replay.body, executor.StartCount()) + } +} + +func TestHubTaskReplay_ConcurrentDifferentDeliveryWaitsForAdmission(t *testing.T) { + executor := &admissionExecutor{blockNext: true, entered: make(chan string, 1), release: make(chan struct{})} + server, h := newDeliveryTestServer(t, executor, nil) + defer server.Close() + defer func() { + select { + case <-executor.release: + default: + close(executor.release) + } + }() + first := make(chan postResult, 1) + failures := make(chan error, 1) + go func() { + result, err := httpPostJSON(server.URL, admissionRunBody(h.WorkspaceAllowlist[0], "delivery-inflight", "task-shared", nil)) + if err != nil { + failures <- err + return + } + first <- result + }() + select { + case <-executor.entered: + case err := <-failures: + t.Fatal(err) + case <-time.After(5 * time.Second): + t.Fatal("executor admission did not start") + } + for _, deliveryID := range []string{"different-delivery", ""} { + result := postRunsRaw(t, server.URL, admissionRunBody(h.WorkspaceAllowlist[0], deliveryID, "task-shared", nil)) + if result.status != http.StatusServiceUnavailable || errCode(result.body) != "delivery_busy" { + t.Fatalf("in-flight task replay falsely accepted: %d %#v", result.status, result.body) + } + if result.resp.Header.Get("Retry-After") == "" { + t.Fatal("busy outcome missing retry hint") + } + } + close(executor.release) + var accepted postResult + select { + case accepted = <-first: + case err := <-failures: + t.Fatal(err) + case <-time.After(5 * time.Second): + t.Fatal("first admission did not complete") + } + if accepted.status != http.StatusAccepted { + t.Fatalf("first result: %d %#v", accepted.status, accepted.body) + } + replay := postRunsRaw(t, server.URL, admissionRunBody(h.WorkspaceAllowlist[0], "different-delivery", "task-shared", nil)) + if replay.status != http.StatusAccepted || unwrapSuccess(replay.body)["runId"] != unwrapSuccess(accepted.body)["runId"] || executor.StartCount() != 1 { + t.Fatalf("completed replay mismatch: %#v starts=%d", replay.body, executor.StartCount()) + } +} + +func TestHubTaskReplay_CapacityRejectionCanRetryButAcceptedFailureCannot(t *testing.T) { + executor := &admissionExecutor{failNext: true, err: lifecycle.ErrTooManyConcurrentRuns} + server, h := newDeliveryTestServer(t, executor, nil) + defer server.Close() + body := admissionRunBody(h.WorkspaceAllowlist[0], "delivery-capacity", "task-capacity", nil) + rejected := postRunsRaw(t, server.URL, body) + if rejected.status != http.StatusTooManyRequests || errCode(rejected.body) != "too_many_concurrent_runs" || rejected.resp.Header.Get("Retry-After") == "" { + t.Fatalf("capacity rejection: %d %#v", rejected.status, rejected.body) + } + accepted := postRunsRaw(t, server.URL, body) + if accepted.status != http.StatusAccepted || executor.StartCount() != 2 { + t.Fatalf("capacity retry was not really admitted: %d %#v starts=%d", accepted.status, accepted.body, executor.StartCount()) + } + runID := unwrapSuccess(accepted.body)["runId"].(string) + ensureStore(h).SetRunStatus(runID, "failed") + h.DeliveryDedup = deliverydedup.New(deliverydedup.DefaultCapacity, deliverydedup.DefaultTTL) + replay := postRunsRaw(t, server.URL, body) + if replay.status != http.StatusAccepted || unwrapSuccess(replay.body)["runId"] != runID || unwrapSuccess(replay.body)["status"] != "failed" || executor.StartCount() != 2 { + t.Fatalf("execution failure must not restart accepted work: %d %#v starts=%d", replay.status, replay.body, executor.StartCount()) + } +} + +func TestHubTaskReplay_UncertainAdmissionExposesReadOnlyEvidence(t *testing.T) { + executor := &admissionExecutor{} + server, h := newDeliveryTestServer(t, executor, nil) + defer server.Close() + repository := ensureStore(h) + run, err := repository.CreateRunAdmission("run-needs-review", "proj_local", "thread_local", "task-needs-review") + if err != nil { + t.Fatal(err) + } + result := postRunsRaw(t, server.URL, admissionRunBody(h.WorkspaceAllowlist[0], "delivery-unknown", "task-needs-review", nil)) + if result.status != http.StatusConflict || errCode(result.body) != "admission_uncertain" { + t.Fatalf("orphaned pending admission=%d %#v", result.status, result.body) + } + if executor.StartCount() != 0 { + t.Fatal("uncertain replay started an executor") + } + response, err := http.Get(server.URL + "/v1/runs/" + run.ID) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + var body map[string]any + if err := json.NewDecoder(response.Body).Decode(&body); err != nil { + t.Fatal(err) + } + data := unwrapSuccess(body) + if response.StatusCode != http.StatusOK || data["runId"] != run.ID || data["admissionState"] != store.RunAdmissionPending { + t.Fatalf("GET run must expose admission evidence for reconciliation: %d %#v", response.StatusCode, body) + } +} diff --git a/edge-server/internal/errcode/codes.go b/edge-server/internal/errcode/codes.go index 7fa5665b5..452a87b13 100644 --- a/edge-server/internal/errcode/codes.go +++ b/edge-server/internal/errcode/codes.go @@ -56,9 +56,11 @@ var ( ErrTooManyConcurrentRuns = New("too_many_concurrent_runs", "too many concurrent runs", http.StatusTooManyRequests) // Run - ErrActiveRunExists = New("active_run_exists", "thread already has an active run", http.StatusConflict) - ErrDeliveryBusy = New("delivery_busy", "delivery admission is busy; retry later", http.StatusServiceUnavailable) - ErrDeliveryConflict = New("delivery_conflict", "delivery id belongs to another task or legacy scope", http.StatusConflict) + ErrActiveRunExists = New("active_run_exists", "thread already has an active run", http.StatusConflict) + ErrDeliveryBusy = New("delivery_busy", "delivery admission is busy; retry later", http.StatusServiceUnavailable) + ErrDeliveryConflict = New("delivery_conflict", "delivery id belongs to another task or legacy scope", http.StatusConflict) + ErrAdmissionPersistFailed = New("admission_persist_failed", "run admission evidence could not be persisted; retry later", http.StatusServiceUnavailable) + ErrAdmissionUncertain = New("admission_uncertain", "run admission outcome requires reconciliation; do not restart automatically", http.StatusConflict) // Agent discovery ErrInvalidAgentID = New("invalid_agent_id", "unknown agent adapter", http.StatusBadRequest) diff --git a/edge-server/internal/lifecycle/mock_executor.go b/edge-server/internal/lifecycle/mock_executor.go index 9c7fcccd2..115845c09 100644 --- a/edge-server/internal/lifecycle/mock_executor.go +++ b/edge-server/internal/lifecycle/mock_executor.go @@ -245,6 +245,12 @@ func RunResponse(run store.Run) map[string]any { if run.WorkDir != "" { payload["workDir"] = run.WorkDir } + if run.AdmissionState != "" { + payload["admissionState"] = run.AdmissionState + } + if run.AdmissionErrorCode != "" { + payload["admissionErrorCode"] = run.AdmissionErrorCode + } return payload } diff --git a/edge-server/internal/lifecycle/thread_transcript_test.go b/edge-server/internal/lifecycle/thread_transcript_test.go index f37618bb0..97d5ac13f 100644 --- a/edge-server/internal/lifecycle/thread_transcript_test.go +++ b/edge-server/internal/lifecycle/thread_transcript_test.go @@ -28,6 +28,12 @@ func (w *stubTranscriptWriter) DeleteThread(id string) bool { return false } func (w *stubTranscriptWriter) CreateRun(id, projectID, threadID string) (store.Run, error) { return store.Run{}, nil } +func (w *stubTranscriptWriter) CreateRunAdmission(id, projectID, threadID, hubTaskID string) (store.Run, error) { + return store.Run{}, store.ErrNotFound +} +func (w *stubTranscriptWriter) RecordRunAdmission(id, errorCode string) (store.Run, error) { + return store.Run{}, store.ErrNotFound +} func (w *stubTranscriptWriter) SetRunStatus(id, status string) (store.Run, bool) { return store.Run{}, false } diff --git a/edge-server/internal/runcontrol/admission.go b/edge-server/internal/runcontrol/admission.go new file mode 100644 index 000000000..5b2213579 --- /dev/null +++ b/edge-server/internal/runcontrol/admission.go @@ -0,0 +1,143 @@ +package runcontrol + +import ( + "errors" + "log/slog" + + "github.com/agenthub/edge-server/internal/errcode" + "github.com/agenthub/edge-server/internal/lifecycle" + "github.com/agenthub/edge-server/internal/store" +) + +// Owned only while Create is deciding executor admission. The shared mutex +// protects this transient ownership evidence; no completed receipts accumulate. +var pendingRunAdmissions = make(map[string]struct{}) + +func finishRunAdmission(runID string) { + runCreationMu.Lock() + delete(pendingRunAdmissions, runID) + runCreationMu.Unlock() +} + +// prepareRunAdmission is called under runCreationMu. It atomically chooses a +// retained receipt or creates durable pending identity before executor startup. +func prepareRunAdmission(repository store.Repository, executor lifecycle.RunExecutor, params CreateParams) (store.Run, bool, *errcode.Error) { + if params.HubTaskID != "" { + if existing, found := repository.GetRunByHubTaskID(params.HubTaskID); found { + if err := authorizeRunReplay(existing, params); err != nil { + return store.Run{}, false, err + } + retry, err := replayRunAdmission(repository, existing) + if err != nil { + return store.Run{}, false, err + } + if !retry { + slog.Info("run.dedup", "hubTaskId", params.HubTaskID, "existingRunId", existing.ID) + return existing, true, nil + } + } + } + if params.Cleanup { + cleanupRuns(repository) + } + if err := validateTarget(repository, params.ProjectID, params.ThreadID); err != nil { + return store.Run{}, false, err + } + if err := validateWorkDir(params.WorkDir, params.WorkspaceAllowlist); err != nil { + return store.Run{}, false, err + } + if err := validatePermissionMode(params.PermissionMode); err != nil { + return store.Run{}, false, errcode.ErrInvalidPermissionMode + } + if active, ok := ActiveRunForThread(repository.ListRuns(params.ThreadID)); ok { + return store.Run{}, false, errcode.ErrActiveRunExists.WithMessagef("thread already has an active run: %s", active.ID) + } + if executor == nil { + return store.Run{}, false, errcode.ErrExecutorUnavailable + } + if params.AgentID != "" && params.AgentExists != nil && !params.AgentExists(params.AgentID) { + return store.Run{}, false, errcode.ErrInvalidAgentID.WithMessagef("unknown agent adapter: %q", params.AgentID) + } + if params.HubTaskID != "" && params.BuildContext == nil { + return store.Run{}, false, errcode.ErrExecutorUnavailable.WithMessage("Hub task admission requires an executor context") + } + runID := generateRunID() + var run store.Run + var err error + if params.HubTaskID == "" { + run, err = repository.CreateRun(runID, params.ProjectID, params.ThreadID) + } else { + run, err = repository.CreateRunAdmission(runID, params.ProjectID, params.ThreadID, params.HubTaskID) + } + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return store.Run{}, false, errcode.ErrNotFound.WithMessage("project or thread not found") + } + if params.HubTaskID != "" { + // This attempt has not called Start. Keep that definite rejection in + // memory even if the write path is down; a retry in this process may + // persist it before trying again. A recovered pending record stays unknown. + if retained, ok := repository.GetRun(runID); ok && retained.HubTaskID == params.HubTaskID { + _, _ = repository.RecordRunAdmission(runID, errcode.ErrAdmissionPersistFailed.Code) + repository.SetRunStatusIf(runID, "failed", "queued") + } + return store.Run{}, false, errcode.ErrAdmissionPersistFailed + } + return store.Run{}, false, errcode.ErrInternal.WithMessage("failed to create run") + } + if params.HubTaskID != "" { + pendingRunAdmissions[run.ID] = struct{}{} + } + return run, false, nil +} + +func authorizeRunReplay(run store.Run, params CreateParams) *errcode.Error { + if params.AuthorizeReplay != nil { + return params.AuthorizeReplay(run) + } + if run.ProjectID != params.ProjectID || run.ThreadID != params.ThreadID { + return errcode.ErrDeliveryConflict + } + return nil +} + +// replayRunAdmission returns retry=true only with durable proof of a rejected +// pre-execution attempt. Execution status alone never grants permission to start. +func replayRunAdmission(repository store.Repository, run store.Run) (retry bool, result *errcode.Error) { + if _, owned := pendingRunAdmissions[run.ID]; owned { + return false, errcode.ErrDeliveryBusy + } + switch run.AdmissionState { + case store.RunAdmissionAccepted: + // Also retries a previous final persistence failure without starting work. + if _, err := repository.RecordRunAdmission(run.ID, ""); err != nil { + return false, errcode.ErrAdmissionPersistFailed + } + return false, nil + case store.RunAdmissionRejected: + if _, err := repository.RecordRunAdmission(run.ID, run.AdmissionErrorCode); err != nil { + return false, errcode.ErrAdmissionPersistFailed + } + switch run.AdmissionErrorCode { + case errcode.ErrTooManyConcurrentRuns.Code, errcode.ErrAdmissionPersistFailed.Code: + // The former was rejected before executor ownership; the latter was + // rejected before Start was called. A completed/started run is never reset. + if run.StartedAt != "" || (run.Status != "failed" && run.Status != "queued") { + break + } + if _, ok := repository.SetRunStatusIf(run.ID, "failed", "queued", "failed"); !ok { + return false, errcode.ErrAdmissionPersistFailed + } + return true, nil + case errcode.ErrExecutorStartFailed.Code: + return false, errcode.ErrExecutorStartFailed + } + case "": + // Compatibility: a legacy durable started timestamp is positive execution + // evidence. Legacy queued/failed records without it are ambiguous. + if run.StartedAt != "" { + return false, nil + } + } + return false, errcode.ErrAdmissionUncertain.WithMessagef("admission outcome for run %s requires reconciliation; automatic restart is unsafe", run.ID) +} diff --git a/edge-server/internal/runcontrol/admission_test.go b/edge-server/internal/runcontrol/admission_test.go new file mode 100644 index 000000000..e624d195f --- /dev/null +++ b/edge-server/internal/runcontrol/admission_test.go @@ -0,0 +1,143 @@ +package runcontrol + +import ( + "errors" + "path/filepath" + "testing" + + "github.com/agenthub/edge-server/internal/errcode" + "github.com/agenthub/edge-server/internal/store" +) + +type admissionWriteFailure struct { + store.Repository + failPrepare bool + failAccepted bool +} + +func (r *admissionWriteFailure) CreateRunAdmission(id, project, thread, task string) (store.Run, error) { + run, err := r.Repository.CreateRunAdmission(id, project, thread, task) + if err == nil && r.failPrepare { + r.failPrepare = false + return run, errors.New("fixture pending write failure") + } + return run, err +} +func (r *admissionWriteFailure) RecordRunAdmission(id, code string) (store.Run, error) { + run, err := r.Repository.RecordRunAdmission(id, code) + if err == nil && code == "" && r.failAccepted { + r.failAccepted = false + return run, errors.New("fixture accepted write failure") + } + return run, err +} + +func TestHubAdmission_PersistenceFailuresNeverAcknowledgeOrDuplicate(t *testing.T) { + for _, phase := range []string{"before-executor", "after-executor"} { + t.Run(phase, func(t *testing.T) { + repo := &admissionWriteFailure{Repository: newTestRepo(t), failPrepare: phase == "before-executor", failAccepted: phase == "after-executor"} + executor := &recordingExecutor{} + params := baseParams(t.TempDir()) + params.HubTaskID = "hub-persistence" + if _, err := Create(repo, executor, nil, params); !errors.Is(err, errcode.ErrAdmissionPersistFailed) { + t.Fatalf("write failure err=%v", err) + } + wantStarts := 0 + if phase == "after-executor" { + wantStarts = 1 + } + if executor.startCount() != wantStarts { + t.Fatalf("start count after failure=%d want=%d", executor.startCount(), wantStarts) + } + run, err := Create(repo, executor, nil, params) + if err != nil || run.AdmissionState != store.RunAdmissionAccepted || executor.startCount() != 1 { + t.Fatalf("retry=%#v err=%v starts=%d", run, err, executor.startCount()) + } + if _, err := Create(repo, executor, nil, params); err != nil || executor.startCount() != 1 { + t.Fatalf("replay err=%v starts=%d", err, executor.startCount()) + } + }) + } +} + +func TestHubAdmission_ReopenKeepsIdentityWithoutRestarting(t *testing.T) { + constructors := map[string]func(string) (store.Repository, error){ + "file": func(p string) (store.Repository, error) { return store.NewFile(p) }, + "sqlite": func(p string) (store.Repository, error) { return store.NewSQLite(p) }, + } + for backend, open := range constructors { + t.Run(backend, func(t *testing.T) { + for _, phase := range []string{"accepted", "pending", "legacy-queued", "legacy-started"} { + t.Run(phase, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "runs.db") + repo, err := open(path) + if err != nil { + t.Fatal(err) + } + if _, err = repo.CreateProject("proj_local", "Local", ""); err != nil { + t.Fatal(err) + } + if _, err = repo.CreateThread("thread_local", "proj_local", "Local", "direct", "", ""); err != nil { + t.Fatal(err) + } + params := baseParams(t.TempDir()) + params.HubTaskID = "hub-reopen" + executor := &recordingExecutor{} + var original store.Run + switch phase { + case "accepted": + original, err = Create(repo, executor, nil, params) + case "pending": + original, err = repo.CreateRunAdmission("run-pending", "proj_local", "thread_local", params.HubTaskID) + default: + original, err = repo.CreateRun("run-legacy", "proj_local", "thread_local") + if err == nil { + original, _ = repo.SetRunHubTaskID(original.ID, params.HubTaskID) + } + if phase == "legacy-started" { + original, _ = repo.SetRunStatus(original.ID, "started") + } + } + if err != nil { + t.Fatal(err) + } + repo.Close() + recovered, err := open(path) + if err != nil { + t.Fatal(err) + } + defer recovered.Close() + afterRestart := &recordingExecutor{} + run, replayErr := Create(recovered, afterRestart, nil, params) + if phase == "pending" || phase == "legacy-queued" { + if !errors.Is(replayErr, errcode.ErrAdmissionUncertain) { + t.Fatalf("ambiguous record err=%v run=%#v", replayErr, run) + } + } else if replayErr != nil || run.ID != original.ID { + t.Fatalf("retained accepted identity=%#v err=%v", run, replayErr) + } + if afterRestart.startCount() != 0 { + t.Fatal("reopening storage must not restart any retained run") + } + }) + } + }) + } +} + +func TestHubAdmission_DefaultReplayCannotCrossScope(t *testing.T) { + repo := newTestRepo(t) + executor := &recordingExecutor{} + params := baseParams(t.TempDir()) + params.HubTaskID = "hub-scope" + if _, err := Create(repo, executor, nil, params); err != nil { + t.Fatal(err) + } + params.ProjectID = "another-project" + if _, err := Create(repo, executor, nil, params); !errors.Is(err, errcode.ErrDeliveryConflict) { + t.Fatalf("unscoped replay err=%v", err) + } + if executor.startCount() != 1 { + t.Fatal("conflicting replay started another executor") + } +} diff --git a/edge-server/internal/runcontrol/runcontrol.go b/edge-server/internal/runcontrol/runcontrol.go index 3c50ad7e7..1edca18d1 100644 --- a/edge-server/internal/runcontrol/runcontrol.go +++ b/edge-server/internal/runcontrol/runcontrol.go @@ -29,8 +29,8 @@ import ( // check-then-create sequence (no active run on the thread, then CreateRun); // without one shared lock, two concurrent requests — including one from REST // and one from MCP — could both pass the active-run check and create -// overlapping runs. Run creation is rare and cheap, so the contention cost of -// a process-wide lock is negligible compared to the invariant it protects. +// overlapping runs. For Hub work this section includes the durable pending +// identity write. Timeline publication and executor startup stay outside it. var runCreationMu sync.Mutex const ( @@ -73,11 +73,16 @@ type CreateParams struct { // prompt and queued-marker items; MCP publishes a single user_message item. Timeline func(run store.Run) - // HubTaskID is the Hub-side task ID for deduplication. When non-empty, - // Create returns the existing run if one with the same HubTaskID already - // exists (idempotent redelivery guard for orphan recovery). + // HubTaskID identifies one logical Hub task across delivery transports. + // A retained run is replayable only with admission evidence, not merely + // because a queued/failed record exists. HubTaskID string + // AuthorizeReplay validates the actual stored scope before replaying a Hub + // task. Transports with capability policy must supply it. Without a policy, + // replay is restricted to the same project/thread as the request. + AuthorizeReplay func(store.Run) *errcode.Error + // BuildContext builds the RunProcessContext handed to the executor. // When nil, the executor start step is skipped. BuildContext func(run store.Run) lifecycle.RunProcessContext @@ -95,60 +100,17 @@ func Create(repository store.Repository, executor lifecycle.RunExecutor, bus *ev } params.WorkDir = strings.TrimSpace(params.WorkDir) - // The lock covers only the check-then-create section (matching the - // historical PostRuns lock scope): cleanup, validation, and CreateRun. - // Event publication, timeline hooks, and the executor start run outside - // the lock so slow executor starts never serialize behind each other. runCreationMu.Lock() - if params.Cleanup { - cleanupRuns(repository) - } - if err := validateTarget(repository, params.ProjectID, params.ThreadID); err != nil { - runCreationMu.Unlock() - return store.Run{}, err - } - if err := validateWorkDir(params.WorkDir, params.WorkspaceAllowlist); err != nil { - runCreationMu.Unlock() + run, replayed, err := prepareRunAdmission(repository, executor, params) + runCreationMu.Unlock() + if err != nil { return store.Run{}, err } - if err := validatePermissionMode(params.PermissionMode); err != nil { - runCreationMu.Unlock() - slog.Error("invalid permission mode", "permissionMode", params.PermissionMode, "error", err) - return store.Run{}, errcode.ErrInvalidPermissionMode + if replayed { + return run, nil } - // HubTaskID dedup: when a non-empty HubTaskID matches an existing run, - // return that run idempotently instead of creating a duplicate. This - // guards against orphan-recovery redelivery races (issue #2066). if params.HubTaskID != "" { - if existing, found := repository.GetRunByHubTaskID(params.HubTaskID); found { - runCreationMu.Unlock() - slog.Info("run.dedup", "hubTaskId", params.HubTaskID, "existingRunId", existing.ID) - return existing, nil - } - } - if active, ok := ActiveRunForThread(repository.ListRuns(params.ThreadID)); ok { - runCreationMu.Unlock() - return store.Run{}, errcode.ErrActiveRunExists.WithMessagef("thread already has an active run: %s", active.ID) - } - if executor == nil { - runCreationMu.Unlock() - return store.Run{}, errcode.ErrExecutorUnavailable.WithMessage("no Agent Runtime executor configured") - } - // #175: Reject unknown agentId — do not fall back to default adapter. - if params.AgentID != "" && params.AgentExists != nil && !params.AgentExists(params.AgentID) { - runCreationMu.Unlock() - return store.Run{}, errcode.ErrInvalidAgentID.WithMessagef("unknown agent adapter: %q", params.AgentID) - } - run, err := repository.CreateRun(generateRunID(), params.ProjectID, params.ThreadID) - if err == nil && params.HubTaskID != "" { - run, _ = repository.SetRunHubTaskID(run.ID, params.HubTaskID) - } - runCreationMu.Unlock() - if err != nil { - if errors.Is(err, store.ErrNotFound) { - return store.Run{}, errcode.ErrNotFound.WithMessage("project or thread not found") - } - return store.Run{}, errcode.ErrInternal.WithMessagef("failed to create run: %v", err) + defer finishRunAdmission(run.ID) } scope := map[string]any{ @@ -175,12 +137,24 @@ func Create(repository store.Repository, executor lifecycle.RunExecutor, bus *ev "error": "run execution failed", }) } + admissionErr := errcode.ErrExecutorStartFailed if errors.Is(err, lifecycle.ErrTooManyConcurrentRuns) { - slog.Error("too many concurrent runs", "runId", run.ID, "error", err) - return store.Run{}, errcode.ErrTooManyConcurrentRuns + admissionErr = errcode.ErrTooManyConcurrentRuns } - return store.Run{}, errcode.ErrExecutorStartFailed + if params.HubTaskID != "" { + if _, persistErr := repository.RecordRunAdmission(run.ID, admissionErr.Code); persistErr != nil { + return store.Run{}, errcode.ErrAdmissionPersistFailed + } + } + return store.Run{}, admissionErr + } + } + if params.HubTaskID != "" { + accepted, persistErr := repository.RecordRunAdmission(run.ID, "") + if persistErr != nil { + return store.Run{}, errcode.ErrAdmissionPersistFailed } + return accepted, nil } return run, nil } diff --git a/edge-server/internal/store/admission_cleanup_test.go b/edge-server/internal/store/admission_cleanup_test.go new file mode 100644 index 000000000..9a411f89b --- /dev/null +++ b/edge-server/internal/store/admission_cleanup_test.go @@ -0,0 +1,32 @@ +package store + +import ( + "testing" + "time" +) + +func TestCleanupRuns_RetainsPendingAdmissionAfterEarlyExecutionFinish(t *testing.T) { + repo := New() + if _, err := repo.CreateProject("p", "Project", ""); err != nil { + t.Fatal(err) + } + if _, err := repo.CreateThread("t", "p", "Thread", "direct", "", ""); err != nil { + t.Fatal(err) + } + for _, id := range []string{"pending", "accepted"} { + if _, err := repo.CreateRunAdmission(id, "p", "t", id); err != nil { + t.Fatal(err) + } + repo.SetRunStatus(id, "finished") + } + if _, err := repo.RecordRunAdmission("accepted", ""); err != nil { + t.Fatal(err) + } + repo.CleanupRuns(RunCleanupOptions{Now: time.Now().Add(48 * time.Hour), TerminalTTL: time.Hour, MaxTerminalRunsPerThread: 1}) + if _, ok := repo.GetRun("pending"); !ok { + t.Fatal("retention removed uncertain admission evidence") + } + if _, ok := repo.GetRun("accepted"); ok { + t.Fatal("ordinary accepted terminal record should still expire") + } +} diff --git a/edge-server/internal/store/file_store.go b/edge-server/internal/store/file_store.go index 983a45d6c..d52ec61b1 100644 --- a/edge-server/internal/store/file_store.go +++ b/edge-server/internal/store/file_store.go @@ -250,6 +250,28 @@ func (f *FileStore) CreateRun(id, projectID, threadID string) (Run, error) { return run, nil } +func (f *FileStore) CreateRunAdmission(id, projectID, threadID, hubTaskID string) (Run, error) { + run, err := f.store.CreateRunAdmission(id, projectID, threadID, hubTaskID) + if err != nil { + return Run{}, err + } + if err := f.syncPersist(); err != nil { + return run, err + } + return run, nil +} + +func (f *FileStore) RecordRunAdmission(id, errorCode string) (Run, error) { + run, err := f.store.RecordRunAdmission(id, errorCode) + if err != nil { + return run, err + } + if err := f.syncPersist(); err != nil { + return run, err + } + return run, nil +} + func (f *FileStore) GetRun(id string) (Run, bool) { return f.store.GetRun(id) } diff --git a/edge-server/internal/store/sqlite_store.go b/edge-server/internal/store/sqlite_store.go index 1413b4c1b..8999c9bca 100644 --- a/edge-server/internal/store/sqlite_store.go +++ b/edge-server/internal/store/sqlite_store.go @@ -376,6 +376,16 @@ func (s *SQLiteStore) CreateRun(id, projectID, threadID string) (Run, error) { return persistAfterSQLiteWrite(s, run, err) } +func (s *SQLiteStore) CreateRunAdmission(id, projectID, threadID, hubTaskID string) (Run, error) { + run, err := s.store.CreateRunAdmission(id, projectID, threadID, hubTaskID) + return persistAfterSQLiteWrite(s, run, err) +} + +func (s *SQLiteStore) RecordRunAdmission(id, errorCode string) (Run, error) { + run, err := s.store.RecordRunAdmission(id, errorCode) + return persistAfterSQLiteWrite(s, run, err) +} + func (s *SQLiteStore) GetRun(id string) (Run, bool) { return s.store.GetRun(id) } diff --git a/edge-server/internal/store/store_domain.go b/edge-server/internal/store/store_domain.go index f46f26ffa..cb993e0a9 100644 --- a/edge-server/internal/store/store_domain.go +++ b/edge-server/internal/store/store_domain.go @@ -335,8 +335,9 @@ func (s *Store) GetRunByHubTaskID(hubTaskID string) (Run, bool) { } s.mu.RLock() defer s.mu.RUnlock() - for _, run := range s.runs { - if run.HubTaskID == hubTaskID { + for i := len(s.runOrder) - 1; i >= 0; i-- { + run, ok := s.runs[s.runOrder[i]] + if ok && run.HubTaskID == hubTaskID { return run, true } } diff --git a/edge-server/internal/store/store_interfaces.go b/edge-server/internal/store/store_interfaces.go index c1486a85f..b53da49bf 100644 --- a/edge-server/internal/store/store_interfaces.go +++ b/edge-server/internal/store/store_interfaces.go @@ -33,6 +33,8 @@ type Writer interface { UpdateThread(id string, title *string, status *string) (Thread, bool) DeleteThread(id string) bool CreateRun(id, projectID, threadID string) (Run, error) + CreateRunAdmission(id, projectID, threadID, hubTaskID string) (Run, error) + RecordRunAdmission(id, errorCode string) (Run, error) SetRunStatus(id, status string) (Run, bool) SetRunStatusIf(id, status string, allowedCurrent ...string) (Run, bool) CreateItem(item Item) (Item, error) diff --git a/edge-server/internal/store/store_query_plan.go b/edge-server/internal/store/store_query_plan.go index d84db63a4..a26d6f0c2 100644 --- a/edge-server/internal/store/store_query_plan.go +++ b/edge-server/internal/store/store_query_plan.go @@ -315,7 +315,9 @@ func buildTerminalCleanupCandidates(order []string, runs map[string]Run) []runCl candidates := make([]runCleanupCandidate, 0, len(order)) for idx, id := range order { run, ok := runs[id] - if !ok || !isTerminalRunStatus(run.Status) { + // An executor may finish before its admission receipt is committed. + // Pending evidence must survive retention until the outcome is known. + if !ok || run.AdmissionState == RunAdmissionPending || !isTerminalRunStatus(run.Status) { continue } terminalAt, hasTime := runTerminalTime(run) diff --git a/edge-server/internal/store/store_run_admission.go b/edge-server/internal/store/store_run_admission.go new file mode 100644 index 000000000..a2dd4ba06 --- /dev/null +++ b/edge-server/internal/store/store_run_admission.go @@ -0,0 +1,102 @@ +package store + +import ( + "errors" + "fmt" +) + +// Run admission lifecycle states persisted alongside Hub task identity. +const ( + RunAdmissionPending = "pending" + RunAdmissionAccepted = "accepted" + RunAdmissionRejected = "rejected" +) + +var ( + ErrRunAdmissionHubTaskIDRequired = errors.New("run admission hub task id is required") + ErrRunAdmissionInvalidTransition = errors.New("invalid run admission transition") + ErrRunAdmissionExists = errors.New("run admission already exists") +) + +// CreateRunAdmission creates a run and records its Hub task identity and pending +// admission atomically under one Store lock. It reuses the existing run create +// validation/order helper; only the non-empty HubTaskID and admission marker are new. +func (s *Store) CreateRunAdmission(id, projectID, threadID, hubTaskID string) (Run, error) { + if hubTaskID == "" { + return Run{}, ErrRunAdmissionHubTaskIDRequired + } + s.mu.Lock() + defer s.mu.Unlock() + + _, existed := s.runs[id] + run, order, err := createRunInMaps(s.projects, s.threads, s.runs, s.runOrder, id, projectID, threadID, nowString()) + if err != nil { + return Run{}, err + } + if existed { + if run.AdmissionState == RunAdmissionPending && + run.HubTaskID == hubTaskID && + run.ProjectID == projectID && + run.ThreadID == threadID { + return run, nil + } + return Run{}, admissionCreateConflictError(id, hubTaskID, projectID, threadID, run) + } + + s.runOrder = order + run.HubTaskID = hubTaskID + run.AdmissionState = RunAdmissionPending + run.AdmissionErrorCode = "" + s.runs[id] = run + return run, nil +} + +// RecordRunAdmission records the final executor admission outcome. An empty +// errorCode means accepted; a non-empty code means rejected and is retained as +// the public-safe error identity. Pending is the only pre-final state. Repeating +// the same final outcome is idempotent; any other transition is rejected. This +// method intentionally mutates only the admission fields, never execution metadata. +func (s *Store) RecordRunAdmission(id, errorCode string) (Run, error) { + s.mu.Lock() + defer s.mu.Unlock() + + run, ok := s.runs[id] + if !ok { + return Run{}, ErrNotFound + } + nextState := RunAdmissionAccepted + if errorCode != "" { + nextState = RunAdmissionRejected + } + switch run.AdmissionState { + case RunAdmissionPending: + run.AdmissionState = nextState + if nextState == RunAdmissionAccepted { + run.AdmissionErrorCode = "" + } else { + run.AdmissionErrorCode = errorCode + } + s.runs[id] = run + return run, nil + case RunAdmissionAccepted: + if nextState != RunAdmissionAccepted || errorCode != "" { + return run, admissionTransitionError(run.AdmissionState, nextState, errorCode) + } + return run, nil + case RunAdmissionRejected: + if nextState != RunAdmissionRejected || run.AdmissionErrorCode != errorCode { + return run, admissionTransitionError(run.AdmissionState, nextState, errorCode) + } + return run, nil + default: + return run, admissionTransitionError(run.AdmissionState, nextState, errorCode) + } +} + +func admissionTransitionError(from, to, errorCode string) error { + return fmt.Errorf("%w: run admission transition %q -> %q with error code %q", ErrRunAdmissionInvalidTransition, from, to, errorCode) +} + +func admissionCreateConflictError(id, hubTaskID, projectID, threadID string, existing Run) error { + return fmt.Errorf("%w: %w: run %q already has admission identity %q in project/thread %q/%q; requested %q in %q/%q", ErrRunAdmissionInvalidTransition, ErrRunAdmissionExists, id, existing.HubTaskID, existing.ProjectID, existing.ThreadID, hubTaskID, projectID, threadID) +} diff --git a/edge-server/internal/store/store_run_admission_test.go b/edge-server/internal/store/store_run_admission_test.go new file mode 100644 index 000000000..d431fa252 --- /dev/null +++ b/edge-server/internal/store/store_run_admission_test.go @@ -0,0 +1,462 @@ +package store + +import ( + "database/sql" + "encoding/json" + "errors" + "os" + "path/filepath" + "testing" +) + +func TestRunAdmissionStateMachine(t *testing.T) { + s := New() + project, thread := seedRunAdmissionStore(t, s) + + if _, err := s.CreateRunAdmission("admission_missing_hub", project.ID, thread.ID, ""); !errors.Is(err, ErrRunAdmissionHubTaskIDRequired) { + t.Fatalf("CreateRunAdmission empty HubTaskID error = %v, want ErrRunAdmissionHubTaskIDRequired", err) + } + + run, err := s.CreateRunAdmission("admission_run_1", project.ID, thread.ID, "hub-task-1") + if err != nil { + t.Fatalf("CreateRunAdmission returned error: %v", err) + } + if run.ID != "admission_run_1" || run.Status != "queued" || run.HubTaskID != "hub-task-1" || run.AdmissionState != RunAdmissionPending || run.AdmissionErrorCode != "" { + t.Fatalf("CreateRunAdmission = %#v, want queued/pending/hub-task-1", run) + } + + if _, ok := s.SetRunStatus(run.ID, "started"); !ok { + t.Fatal("SetRunStatus started returned false") + } + if _, ok := s.SetRunEvidenceGate(run.ID, `{"ok":true}`); !ok { + t.Fatal("SetRunEvidenceGate returned false") + } + if _, ok := s.SetRunRetryCount(run.ID, 3); !ok { + t.Fatal("SetRunRetryCount returned false") + } + if _, ok := s.SetRunWorkDir(run.ID, "run-workdir"); !ok { + t.Fatal("SetRunWorkDir returned false") + } + + before, ok := s.GetRun(run.ID) + if !ok { + t.Fatal("GetRun after state setup returned false") + } + accepted, err := s.RecordRunAdmission(run.ID, "") + if err != nil { + t.Fatalf("RecordRunAdmission accepted returned error: %v", err) + } + if accepted.AdmissionState != RunAdmissionAccepted || accepted.AdmissionErrorCode != "" { + t.Fatalf("RecordRunAdmission accepted = %#v", accepted) + } + if accepted.Status != before.Status || accepted.StartedAt != before.StartedAt || accepted.FinishedAt != before.FinishedAt || + accepted.EvidenceGateResult != before.EvidenceGateResult || accepted.RetryCount != before.RetryCount || accepted.WorkDir != before.WorkDir || + accepted.HubTaskID != before.HubTaskID { + t.Fatalf("RecordRunAdmission changed execution metadata: before=%#v after=%#v", before, accepted) + } + + repeated, err := s.RecordRunAdmission(run.ID, "") + if err != nil { + t.Fatalf("RecordRunAdmission repeated accepted returned error: %v", err) + } + if repeated.AdmissionState != RunAdmissionAccepted || repeated.AdmissionErrorCode != "" { + t.Fatalf("repeated accepted = %#v", repeated) + } + if _, err := s.RecordRunAdmission(run.ID, "capacity"); !errors.Is(err, ErrRunAdmissionInvalidTransition) { + t.Fatalf("accepted -> rejected error = %v, want ErrRunAdmissionInvalidTransition", err) + } + + rejected, err := s.CreateRunAdmission("admission_run_2", project.ID, thread.ID, "hub-task-2") + if err != nil { + t.Fatalf("CreateRunAdmission second returned error: %v", err) + } + rejected, err = s.RecordRunAdmission(rejected.ID, "capacity") + if err != nil { + t.Fatalf("RecordRunAdmission rejected returned error: %v", err) + } + if rejected.AdmissionState != RunAdmissionRejected || rejected.AdmissionErrorCode != "capacity" { + t.Fatalf("RecordRunAdmission rejected = %#v", rejected) + } + if _, err := s.RecordRunAdmission(rejected.ID, "capacity"); err != nil { + t.Fatalf("RecordRunAdmission repeated rejected returned error: %v", err) + } + if _, err := s.RecordRunAdmission(rejected.ID, "different"); !errors.Is(err, ErrRunAdmissionInvalidTransition) { + t.Fatalf("rejected different code error = %v, want ErrRunAdmissionInvalidTransition", err) + } + + legacy, err := s.CreateRun("admission_legacy", project.ID, thread.ID) + if err != nil { + t.Fatalf("CreateRun legacy returned error: %v", err) + } + if _, err := s.RecordRunAdmission(legacy.ID, ""); !errors.Is(err, ErrRunAdmissionInvalidTransition) { + t.Fatalf("legacy run admission error = %v, want ErrRunAdmissionInvalidTransition", err) + } + + if _, err := s.RecordRunAdmission("admission_missing", ""); !errors.Is(err, ErrNotFound) { + t.Fatalf("RecordRunAdmission missing error = %v, want ErrNotFound", err) + } +} + +func TestCreateRunAdmissionDoesNotResetExistingFinalOrLegacy(t *testing.T) { + s := New() + project, thread := seedRunAdmissionStore(t, s) + + accepted, err := s.CreateRunAdmission("same_admission_id", project.ID, thread.ID, "same-task") + if err != nil { + t.Fatalf("CreateRunAdmission accepted returned error: %v", err) + } + accepted, err = s.RecordRunAdmission(accepted.ID, "") + if err != nil { + t.Fatalf("RecordRunAdmission accepted returned error: %v", err) + } + + // Rebuilding an existing run ID must not downgrade a final admission. + if _, err := s.CreateRunAdmission(accepted.ID, project.ID, thread.ID, "same-task"); err == nil || + (!errors.Is(err, ErrRunAdmissionInvalidTransition) && !errors.Is(err, ErrRunAdmissionExists)) { + t.Fatalf("CreateRunAdmission same accepted ID error = %v, want explicit admission conflict", err) + } + got, ok := s.GetRun(accepted.ID) + if !ok || got.AdmissionState != RunAdmissionAccepted || got.AdmissionErrorCode != "" || + got.HubTaskID != "same-task" || got.ProjectID != project.ID || got.ThreadID != thread.ID { + t.Fatalf("accepted run changed after same-ID CreateRunAdmission: %#v, %v", got, ok) + } + + // A different HubTaskID must never overwrite the existing binding. + if _, err := s.CreateRunAdmission(accepted.ID, project.ID, thread.ID, "different-task"); err == nil || + (!errors.Is(err, ErrRunAdmissionInvalidTransition) && !errors.Is(err, ErrRunAdmissionExists)) { + t.Fatalf("CreateRunAdmission different HubTaskID error = %v, want explicit conflict", err) + } + got, _ = s.GetRun(accepted.ID) + if got.AdmissionState != RunAdmissionAccepted || got.AdmissionErrorCode != "" || got.HubTaskID != "same-task" { + t.Fatalf("different HubTaskID changed accepted binding: %#v", got) + } + + // A different project/thread must also never overwrite the existing binding. + otherProject, err := s.CreateProject("admission_project_other", "Other", "") + if err != nil { + t.Fatalf("CreateProject other returned error: %v", err) + } + otherThread, err := s.CreateThread("admission_thread_other", otherProject.ID, "Other Thread", "", "", "") + if err != nil { + t.Fatalf("CreateThread other returned error: %v", err) + } + if _, err := s.CreateRunAdmission(accepted.ID, otherProject.ID, otherThread.ID, "same-task"); err == nil || + (!errors.Is(err, ErrRunAdmissionInvalidTransition) && !errors.Is(err, ErrRunAdmissionExists)) { + t.Fatalf("CreateRunAdmission different scope error = %v, want explicit conflict", err) + } + got, _ = s.GetRun(accepted.ID) + if got.AdmissionState != RunAdmissionAccepted || got.HubTaskID != "same-task" || + got.ProjectID != project.ID || got.ThreadID != thread.ID { + t.Fatalf("different scope changed accepted run: %#v", got) + } + + // A matching pending run ID is idempotent. + pending, err := s.CreateRunAdmission("pending_admission_id", project.ID, thread.ID, "pending-task") + if err != nil { + t.Fatalf("CreateRunAdmission pending returned error: %v", err) + } + again, err := s.CreateRunAdmission(pending.ID, project.ID, thread.ID, "pending-task") + if err != nil { + t.Fatalf("CreateRunAdmission matching pending retry returned error: %v", err) + } + if again.AdmissionState != RunAdmissionPending || again.HubTaskID != "pending-task" { + t.Fatalf("matching pending retry changed record: %#v", again) + } + + // New attempts keep using a new run ID and remain usable. + attempt, err := s.CreateRunAdmission("new_attempt_id", project.ID, thread.ID, "same-task") + if err != nil { + t.Fatalf("CreateRunAdmission new attempt returned error: %v", err) + } + if attempt.AdmissionState != RunAdmissionPending || attempt.HubTaskID != "same-task" { + t.Fatalf("new attempt = %#v, want pending same-task", attempt) + } + if got, ok := s.GetRunByHubTaskID("same-task"); !ok || got.ID != attempt.ID { + t.Fatalf("GetRunByHubTaskID after new attempt = %#v, %v; want newest %q", got, ok, attempt.ID) + } + + // Legacy runs without admission state cannot be reset or bound by this method. + legacy, err := s.CreateRun("legacy_admission_id", project.ID, thread.ID) + if err != nil { + t.Fatalf("CreateRun legacy returned error: %v", err) + } + if _, err := s.CreateRunAdmission(legacy.ID, project.ID, thread.ID, "legacy-task"); err == nil || + (!errors.Is(err, ErrRunAdmissionInvalidTransition) && !errors.Is(err, ErrRunAdmissionExists)) { + t.Fatalf("CreateRunAdmission legacy ID error = %v, want explicit conflict", err) + } + legacyGot, ok := s.GetRun(legacy.ID) + if !ok || legacyGot.AdmissionState != "" || legacyGot.HubTaskID != "" { + t.Fatalf("legacy run changed: %#v, %v", legacyGot, ok) + } +} + +func TestRunAdmissionPhasesSurviveReopen(t *testing.T) { + for _, kind := range []string{"file", "sqlite"} { + t.Run(kind, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "admission-store.dat") + first := openRunAdmissionStore(t, kind, path) + project, thread := seedRunAdmissionStore(t, first) + + pending, err := first.CreateRunAdmission("admission_pending", project.ID, thread.ID, "task-pending") + if err != nil { + t.Fatalf("CreateRunAdmission pending returned error: %v", err) + } + assertDurableRunAdmission(t, first, path, pending.ID, RunAdmissionPending, "task-pending", "") + accepted, err := first.CreateRunAdmission("admission_accepted", project.ID, thread.ID, "task-accepted") + if err != nil { + t.Fatalf("CreateRunAdmission accepted returned error: %v", err) + } + accepted, err = first.RecordRunAdmission(accepted.ID, "") + if err != nil { + t.Fatalf("RecordRunAdmission accepted returned error: %v", err) + } + assertDurableRunAdmission(t, first, path, accepted.ID, RunAdmissionAccepted, "task-accepted", "") + rejected, err := first.CreateRunAdmission("admission_rejected", project.ID, thread.ID, "task-rejected") + if err != nil { + t.Fatalf("CreateRunAdmission rejected returned error: %v", err) + } + rejected, err = first.RecordRunAdmission(rejected.ID, "capacity") + if err != nil { + t.Fatalf("RecordRunAdmission rejected returned error: %v", err) + } + assertDurableRunAdmission(t, first, path, rejected.ID, RunAdmissionRejected, "task-rejected", "capacity") + first.Close() + + second := openRunAdmissionStore(t, kind, path) + defer second.Close() + + gotPending, ok := second.GetRun(pending.ID) + if !ok || gotPending.HubTaskID != pending.HubTaskID || gotPending.AdmissionState != RunAdmissionPending || gotPending.AdmissionErrorCode != "" { + t.Fatalf("reopen pending = %#v, %v", gotPending, ok) + } + gotAccepted, ok := second.GetRun(accepted.ID) + if !ok || gotAccepted.HubTaskID != accepted.HubTaskID || gotAccepted.AdmissionState != RunAdmissionAccepted || gotAccepted.AdmissionErrorCode != "" { + t.Fatalf("reopen accepted = %#v, %v", gotAccepted, ok) + } + gotRejected, ok := second.GetRun(rejected.ID) + if !ok || gotRejected.HubTaskID != rejected.HubTaskID || gotRejected.AdmissionState != RunAdmissionRejected || gotRejected.AdmissionErrorCode != "capacity" { + t.Fatalf("reopen rejected = %#v, %v", gotRejected, ok) + } + }) + } +} + +func TestRunAdmissionLatestAttempt(t *testing.T) { + for _, kind := range []string{"memory", "file", "sqlite"} { + t.Run(kind, func(t *testing.T) { + var repo Repository + var path string + if kind == "memory" { + repo = New() + } else { + path = filepath.Join(t.TempDir(), "admission-latest.dat") + repo = openRunAdmissionStore(t, kind, path) + } + project, thread := seedRunAdmissionStore(t, repo) + first, err := repo.CreateRunAdmission("admission_latest_1", project.ID, thread.ID, "hub-task-latest") + if err != nil { + t.Fatalf("CreateRunAdmission first returned error: %v", err) + } + if _, err := repo.RecordRunAdmission(first.ID, "capacity"); err != nil { + t.Fatalf("RecordRunAdmission first returned error: %v", err) + } + second, err := repo.CreateRunAdmission("admission_latest_2", project.ID, thread.ID, "hub-task-latest") + if err != nil { + t.Fatalf("CreateRunAdmission second returned error: %v", err) + } + third, err := repo.CreateRunAdmission("admission_latest_3", project.ID, thread.ID, "hub-task-latest") + if err != nil { + t.Fatalf("CreateRunAdmission third returned error: %v", err) + } + if got, ok := repo.GetRunByHubTaskID("hub-task-latest"); !ok || got.ID != third.ID { + t.Fatalf("GetRunByHubTaskID before reopen = %#v, %v; want %q", got, ok, third.ID) + } + if kind != "memory" { + repo.Close() + reopened := openRunAdmissionStore(t, kind, path) + defer reopened.Close() + if got, ok := reopened.GetRunByHubTaskID("hub-task-latest"); !ok || got.ID != third.ID || got.AdmissionState != RunAdmissionPending { + t.Fatalf("GetRunByHubTaskID after reopen = %#v, %v; want latest pending %q", got, ok, third.ID) + } + } else { + if got, ok := repo.GetRunByHubTaskID("hub-task-latest"); !ok || got.ID != third.ID { + t.Fatalf("GetRunByHubTaskID memory = %#v; want latest %q", got, third.ID) + } + } + _ = second + }) + } +} + +func TestRunAdmissionPersistenceFailureRetry(t *testing.T) { + for _, kind := range []string{"file", "sqlite"} { + t.Run(kind, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "admission-persist.dat") + repo := openRunAdmissionStore(t, kind, path) + project, thread := seedRunAdmissionStore(t, repo) + + block, unblock := blockRunAdmissionPersistence(t, repo) + + block() + run, err := repo.CreateRunAdmission("admission_persist_retry", project.ID, thread.ID, "hub-task-persist") + if err == nil { + t.Fatal("CreateRunAdmission with blocked persistence returned nil error") + } + if run.HubTaskID != "hub-task-persist" || run.AdmissionState != RunAdmissionPending { + t.Fatalf("CreateRunAdmission on persist failure run = %#v, want pending marker", run) + } + unblock() + + accepted, err := repo.RecordRunAdmission(run.ID, "") + if err != nil { + t.Fatalf("RecordRunAdmission after recovery returned error: %v", err) + } + if accepted.AdmissionState != RunAdmissionAccepted { + t.Fatalf("RecordRunAdmission after recovery = %#v, want accepted", accepted) + } + + block() + if _, err := repo.RecordRunAdmission(run.ID, ""); err == nil { + t.Fatal("duplicate RecordRunAdmission with blocked persistence returned nil error") + } + unblock() + retried, err := repo.RecordRunAdmission(run.ID, "") + if err != nil { + t.Fatalf("duplicate RecordRunAdmission after recovery returned error: %v", err) + } + if retried.AdmissionState != RunAdmissionAccepted || retried.AdmissionErrorCode != "" { + t.Fatalf("duplicate retry run = %#v", retried) + } + + persistErr, ok := repo.(interface{ LastPersistError() error }) + if !ok { + t.Fatal("Repository does not expose LastPersistError") + } + if err := persistErr.LastPersistError(); err != nil { + t.Fatalf("LastPersistError after successful retry = %v, want nil", err) + } + repo.Close() + }) + } +} + +func seedRunAdmissionStore(t *testing.T, repo Repository) (Project, Thread) { + t.Helper() + project, err := repo.CreateProject("admission_project", "Admission Project", "") + if err != nil { + t.Fatalf("CreateProject returned error: %v", err) + } + thread, err := repo.CreateThread("admission_thread", project.ID, "Admission Thread", "", "", "") + if err != nil { + t.Fatalf("CreateThread returned error: %v", err) + } + return project, thread +} + +func openRunAdmissionStore(t *testing.T, kind, path string) Repository { + t.Helper() + switch kind { + case "file": + store, err := NewFile(path) + if err != nil { + t.Fatalf("NewFile returned error: %v", err) + } + t.Cleanup(store.Close) + return store + case "sqlite": + store, err := NewSQLite(path) + if err != nil { + t.Fatalf("NewSQLite returned error: %v", err) + } + t.Cleanup(store.Close) + return store + default: + t.Fatalf("unknown admission store kind %q", kind) + return nil + } +} + +func assertDurableRunAdmission(t *testing.T, repo Repository, path, runID, state, hubTaskID, errorCode string) { + t.Helper() + var run Run + var ok bool + switch r := repo.(type) { + case *FileStore: + run, ok = readFileSnapshotRun(t, path, runID) + case *SQLiteStore: + run, ok = readSQLiteRunMetadata(t, r, runID) + default: + t.Fatalf("unsupported durable admission repo %T", repo) + } + if !ok { + t.Fatalf("durable run %q not found in %T", runID, repo) + } + if run.HubTaskID != hubTaskID || run.AdmissionState != state || run.AdmissionErrorCode != errorCode { + t.Fatalf("durable run %q = %#v, want HubTaskID=%q state=%q errorCode=%q", runID, run, hubTaskID, state, errorCode) + } +} + +func readFileSnapshotRun(t *testing.T, path, runID string) (Run, bool) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read file snapshot %s: %v", path, err) + } + var snapshot fileSnapshot + if err := json.Unmarshal(data, &snapshot); err != nil { + t.Fatalf("decode file snapshot %s: %v", path, err) + } + run, ok := snapshot.Runs[runID] + return run, ok +} + +// SQLite keeps run admission metadata in the durable row payload. The edge_runs +// relational projection is a separate view and does not carry these fields. +func readSQLiteRunMetadata(t *testing.T, r *SQLiteStore, runID string) (Run, bool) { + t.Helper() + var payload string + err := r.db.QueryRow(`SELECT payload FROM agenthub_store_rows WHERE row_kind = ? AND row_id = ?`, sqliteRowKindRun, runID).Scan(&payload) + if errors.Is(err, sql.ErrNoRows) { + return Run{}, false + } + if err != nil { + t.Fatalf("query sqlite run metadata %s: %v", runID, err) + } + var run Run + if err := decodeSQLiteRowPayload(payload, &run); err != nil { + t.Fatalf("decode sqlite run metadata %s: %v", runID, err) + } + return run, true +} + +func blockRunAdmissionPersistence(t *testing.T, repo Repository) (func(), func()) { + t.Helper() + switch r := repo.(type) { + case *FileStore: + return func() { + if err := os.Remove(r.path); err != nil && !errors.Is(err, os.ErrNotExist) { + t.Fatalf("remove file snapshot before block: %v", err) + } + if err := os.Mkdir(r.path, 0o750); err != nil { + t.Fatalf("mkdir snapshot path to block rename: %v", err) + } + }, func() { + if err := os.Remove(r.path); err != nil { + t.Fatalf("remove snapshot path block: %v", err) + } + } + case *SQLiteStore: + return func() { + if _, err := r.db.Exec(`PRAGMA query_only = ON`); err != nil { + t.Fatalf("enable sqlite query_only: %v", err) + } + }, func() { + if _, err := r.db.Exec(`PRAGMA query_only = OFF`); err != nil { + t.Fatalf("disable sqlite query_only: %v", err) + } + } + default: + t.Fatalf("unsupported persistence repo %T", repo) + return nil, nil + } +} diff --git a/edge-server/internal/store/store_types.go b/edge-server/internal/store/store_types.go index f0007d3cd..4f37ae44a 100644 --- a/edge-server/internal/store/store_types.go +++ b/edge-server/internal/store/store_types.go @@ -39,6 +39,8 @@ type Run struct { EvidenceGateResult string `json:"evidenceGateResult,omitempty"` WorkDir string `json:"workDir,omitempty"` HubTaskID string `json:"hubTaskId,omitempty"` + AdmissionState string `json:"admissionState,omitempty"` + AdmissionErrorCode string `json:"admissionErrorCode,omitempty"` } type RunDiffFile struct { From 5f5f00425e7eb526ebada0396cbe71f3bb635f00 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Mon, 7 Sep 2026 05:49:23 +0800 Subject: [PATCH 2/5] fix(desktop): preserve unconfirmed Hub admission outcomes Treat explicit capacity/persistence failures as recheckable, preserve uncertain outcomes without ACK/FAIL and clear their note only after first acceptance. Extend real-renderer fixtures and keep typed error clearing without a cast. Refs #2349. Co-authored-by: Codex --- .../__e2e__/hub-delivery-admission.spec.ts | 57 ++++++- .../src/__tests__/useHubIntegration.test.ts | 156 ++++++++++++++++++ .../src/hooks/hubIntegrationHelpers.test.ts | 21 +++ .../src/hooks/hubIntegrationMappers.ts | 45 +++-- app/desktop/src/hooks/useHubIntegration.ts | 25 ++- app/desktop/src/stores/taskBridgeStore.ts | 3 +- 6 files changed, 283 insertions(+), 24 deletions(-) diff --git a/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts b/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts index 5a288ac7f..e902ac45c 100644 --- a/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts +++ b/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts @@ -29,6 +29,15 @@ async function readTaskState(page: Page) { }); } +async function readTaskError(page: Page) { + return page.evaluate(async () => { + const modulePath = '/src/stores/taskBridgeStore.ts'; + const { useTaskBridgeStore } = await import(/* @vite-ignore */ modulePath); + const state = useTaskBridgeStore.getState(); + return state.tasks[0]?.error ?? null; + }); +} + async function installDispatchFixture(page: Page, baseURL: string, theme: 'light' | 'dark') { const appOrigin = new URL(baseURL).origin; const hubSockets = new Set(); @@ -41,7 +50,7 @@ async function installDispatchFixture(page: Page, baseURL: string, theme: 'light fails: [] as Record[], registered: 0, targetsRead: 0, - rejection: 503, + rejection: { status: 503, code: 'delivery_busy' } as { status: number; code: string } | null, pageErrors: [] as string[], unhandledWrites: [] as string[], }; @@ -119,8 +128,8 @@ async function installDispatchFixture(page: Page, baseURL: string, theme: 'light } else if (url.pathname === '/v1/runs' && request.method() === 'POST') { calls.runs.push(request.postDataJSON()); if (calls.rejection) { - await route.fulfill({ status: calls.rejection, headers: { 'Retry-After': '1' }, json: { - error: { code: calls.rejection === 503 ? 'delivery_busy' : 'internal_error', message: 'fixture admission rejection', traceId: 'fixture-trace' }, + await route.fulfill({ status: calls.rejection.status, headers: calls.rejection.status === 503 ? { 'Retry-After': '1' } : {}, json: { + error: { code: calls.rejection.code, message: 'fixture admission rejection: ' + calls.rejection.code, traceId: 'fixture-trace' }, } }); } else { await json({ code: 'OK', data: { runId: RUN, projectId: 'proj_local', threadId: THREAD, status: 'queued', deduplicated: calls.runs.length > 2, deliveryId: DELIVERY } }, 202); @@ -192,26 +201,58 @@ for (const theme of ['light', 'dark'] as const) { test('actual Desktop bridge retries admission and repairs lost ACKs (' + theme + ')', async ({ page, baseURL }, testInfo) => { if (!baseURL) throw new Error('Desktop E2E baseURL is required'); const { calls, dispatch, finish } = await installDispatchFixture(page, baseURL, theme); + const queued = { tasks: [{ taskId: TASK, status: 'queued', runId: null }], runToTask: {} }; + await dispatch(); await expect.poll(() => calls.runs.length).toBe(1); - await expect.poll(() => readTaskState(page)).toEqual({ tasks: [{ taskId: TASK, status: 'queued', runId: null }], runToTask: {} }); + await expect.poll(() => readTaskState(page)).toEqual(queued); expect(calls.runs[0]).toMatchObject({ deliveryId: DELIVERY, hubTaskId: TASK, targetId: TARGET, edgeDeviceId: DEVICE }); expect(calls.acks).toHaveLength(0); expect(calls.relayAcks).toHaveLength(0); expect(calls.fails).toHaveLength(0); - calls.rejection = 0; + calls.rejection = { status: 429, code: 'too_many_concurrent_runs' }; + await dispatch(); + await expect.poll(() => calls.runs.length).toBe(2); + await expect.poll(() => readTaskState(page)).toEqual(queued); + expect(calls.acks).toHaveLength(0); + expect(calls.relayAcks).toHaveLength(0); + expect(calls.fails).toHaveLength(0); + + calls.rejection = { status: 503, code: 'admission_persist_failed' }; await dispatch(); + await expect.poll(() => calls.runs.length).toBe(3); + await expect.poll(() => readTaskState(page)).toEqual(queued); + expect(calls.acks).toHaveLength(0); + expect(calls.relayAcks).toHaveLength(0); + expect(calls.fails).toHaveLength(0); + + // admission_uncertain is human-review territory: queued, no ACK/FAIL/relay-ACK. + calls.rejection = { status: 409, code: 'admission_uncertain' }; + await dispatch(); + await expect.poll(() => calls.runs.length).toBe(4); + await expect.poll(() => readTaskState(page)).toEqual(queued); + await expect.poll(() => readTaskError(page)).toContain('fixture admission rejection: admission_uncertain'); + expect(calls.acks).toHaveLength(0); + expect(calls.relayAcks).toHaveLength(0); + expect(calls.fails).toHaveLength(0); + + // The test drives the next dispatch after manual review; this is not an + // automatic client-side retry and the hook itself does not start one. + calls.rejection = null; + await dispatch(); + await expect.poll(() => calls.runs.length).toBe(5); await expect.poll(() => calls.acks.length).toBe(1); await expect.poll(() => calls.relayAcks.length).toBe(1); await expect.poll(() => readTaskState(page)).toEqual({ tasks: [{ taskId: TASK, status: 'running', runId: RUN }], runToTask: { [RUN]: TASK } }); + await expect.poll(() => readTaskError(page)).toBeNull(); // Both first ACK requests failed. A successful replay must send them again. await dispatch(); await expect.poll(() => calls.acks.length).toBe(2); await expect.poll(() => calls.relayAcks.length).toBe(2); expect(calls.acks).toEqual([{ run_id: RUN }, { run_id: RUN }]); - expect(calls.runs).toHaveLength(3); + expect(calls.runs).toHaveLength(6); expect(calls.fails).toHaveLength(0); finish(); @@ -224,9 +265,9 @@ for (const theme of ['light', 'dark'] as const) { await expect.poll(() => readTaskState(page)).toEqual(finished); expect(calls.done).toHaveLength(1); - calls.rejection = 500; + calls.rejection = { status: 500, code: 'internal_error' }; await dispatch(); - await expect.poll(() => calls.runs.length).toBe(5); + await expect.poll(() => calls.runs.length).toBe(8); await expect.poll(() => readTaskState(page)).toEqual(finished); expect(calls.fails).toHaveLength(0); expect(calls.acks).toHaveLength(3); diff --git a/app/desktop/src/__tests__/useHubIntegration.test.ts b/app/desktop/src/__tests__/useHubIntegration.test.ts index 564502044..dc5d18b5f 100644 --- a/app/desktop/src/__tests__/useHubIntegration.test.ts +++ b/app/desktop/src/__tests__/useHubIntegration.test.ts @@ -739,6 +739,106 @@ describe('useHubIntegration', () => { expect(hubClient.failTask).not.toHaveBeenCalled(); }); + it('keeps a too_many_concurrent_runs rejection queued without acking or failing', async () => { + mockRunCreateResponseWithStatus({ error: { code: 'too_many_concurrent_runs', message: 'too many', traceId: 'trace_001' } }, 429); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('queued'); + expect(hoisted.storeTasks[0]?.runId).toBeUndefined(); + expect(hubClient.ackTask).not.toHaveBeenCalled(); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('keeps an admission_persist_failed rejection queued without acking or failing', async () => { + mockRunCreateResponseWithStatus({ error: { code: 'admission_persist_failed', message: 'persist failed', traceId: 'trace_001' } }, 503); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('queued'); + expect(hoisted.storeTasks[0]?.runId).toBeUndefined(); + expect(hubClient.ackTask).not.toHaveBeenCalled(); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('keeps an admission_uncertain rejection queued for manual review without acking or failing', async () => { + mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }, 409); + renderHook(() => + useHubIntegration({ + hubWS, + hubClient, + dispatchTarget: { targetId: 'target-current', deviceId: 'desktop-current' }, + }), + ); + + const relayFrame = { + relay_command_id: 'relay-1', + command_type: 'agent.dispatch', + payload: JSON.stringify( + makeDispatchPayload({ + target_id: 'target-current', + edge_device_id: 'desktop-current', + delivery_id: 'd1', + }), + ), + }; + await act(async () => { + fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('queued'); + expect(hoisted.storeTasks[0]?.runId).toBeUndefined(); + expect(hoisted.storeTasks[0]?.error).toContain('Edge admission result is uncertain'); + expect(hoisted.storeTasks[0]?.error).toContain('manual review'); + expect(hubClient.ackTask).not.toHaveBeenCalled(); + expect(hubClient.failTask).not.toHaveBeenCalled(); + expect(hubClient.ackRelayCommand).not.toHaveBeenCalled(); + }); + + it('clears a queued admission_uncertain error when the same delivery is accepted', async () => { + mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }, 409); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + expect(hoisted.storeTasks[0]?.status).toBe('queued'); + expect(hoisted.storeTasks[0]?.error).toContain('manual review'); + + mockRunSequence('run-1'); + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hoisted.storeTasks[0]?.error).toBeUndefined(); + expect(hubClient.ackTask).toHaveBeenCalledTimes(1); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + it('keeps an existing active run on active_run_exists without acking or failing', async () => { mockRunSequence('run-1'); renderHook(() => useHubIntegration({ hubWS, hubClient })); @@ -766,6 +866,62 @@ describe('useHubIntegration', () => { expect(hubClient.failTask).not.toHaveBeenCalled(); }); + it('does not downgrade an already-running task on admission_uncertain', async () => { + mockRunSequence('run-1'); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + + mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }, 409); + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hoisted.storeTasks[0]?.error).toBeUndefined(); + expect(hubClient.ackTask).toHaveBeenCalledTimes(1); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + + it('does not clear an existing error on a progressed successful replay', async () => { + mockRunSequence('run-1'); + renderHook(() => useHubIntegration({ hubWS, hubClient })); + + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + hoisted.getStoreState().updateTask('task-1', { error: 'existing running error' }); + + mockRunSequence('run-1'); + await act(async () => { + fireHubEvent( + HUB_EVENTS.AGENT_DISPATCH, + makeDispatchPayload({ delivery_id: 'd1' }), + ); + }); + + expect(hoisted.storeTasks[0]?.status).toBe('running'); + expect(hoisted.storeTasks[0]?.runId).toBe('run-1'); + expect(hoisted.storeTasks[0]?.error).toBe('existing running error'); + expect(hubClient.ackTask).toHaveBeenCalledTimes(2); + expect(hubClient.failTask).not.toHaveBeenCalled(); + }); + it('does not ack a relay command on a delivery_busy rejection', async () => { mockRunCreateResponseWithStatus({ error: { code: 'delivery_busy', message: 'busy', traceId: 'trace_001' } }, 503); renderHook(() => diff --git a/app/desktop/src/hooks/hubIntegrationHelpers.test.ts b/app/desktop/src/hooks/hubIntegrationHelpers.test.ts index 0737dbf44..7bd1e06da 100644 --- a/app/desktop/src/hooks/hubIntegrationHelpers.test.ts +++ b/app/desktop/src/hooks/hubIntegrationHelpers.test.ts @@ -18,6 +18,7 @@ import { extractRunOutputBatch, getTeamRouteContext, hasTaskProgressed, + isAdmissionUncertain, isTerminalBridgeTask, isTransientAdmissionRejection, normalizeRouteDecision, @@ -341,12 +342,18 @@ describe('hubIntegrationMappers', () => { it('classifies transient admission rejections from the canonical Edge error envelope', () => { const busyEnvelope = { error: { code: 'delivery_busy', message: 'busy', traceId: 'trace_001' } }; const activeEnvelope = { error: { code: 'active_run_exists', message: 'active', traceId: 'trace_001' } }; + const capacityEnvelope = { error: { code: 'too_many_concurrent_runs', message: 'capacity', traceId: 'trace_001' } }; + const persistEnvelope = { error: { code: 'admission_persist_failed', message: 'persist failed', traceId: 'trace_001' } }; expect(isTransientAdmissionRejection(503, busyEnvelope)).toBe(true); expect(isTransientAdmissionRejection(409, activeEnvelope)).toBe(true); + expect(isTransientAdmissionRejection(429, capacityEnvelope)).toBe(true); + expect(isTransientAdmissionRejection(503, persistEnvelope)).toBe(true); // Raw JSON string form (what the hook passes from runResp.text()). expect(isTransientAdmissionRejection(503, JSON.stringify(busyEnvelope))).toBe(true); expect(isTransientAdmissionRejection(409, JSON.stringify(activeEnvelope))).toBe(true); + expect(isTransientAdmissionRejection(429, JSON.stringify(capacityEnvelope))).toBe(true); + expect(isTransientAdmissionRejection(503, JSON.stringify(persistEnvelope))).toBe(true); // Non-transient / non-matching envelopes keep the existing failure path. expect(isTransientAdmissionRejection(409, { error: { code: 'delivery_conflict', message: 'x', traceId: 't' } })).toBe(false); @@ -354,10 +361,24 @@ describe('hubIntegrationMappers', () => { expect(isTransientAdmissionRejection(503, { error: { code: 'internal', message: 'x', traceId: 't' } })).toBe(false); expect(isTransientAdmissionRejection(503, 'not-json')).toBe(false); expect(isTransientAdmissionRejection(200, {})).toBe(false); + expect(isTransientAdmissionRejection(404, { error: { code: 'too_many_concurrent_runs' } })).toBe(false); + expect(isTransientAdmissionRejection(429, { error: { code: 'unknown' } })).toBe(false); + expect(isTransientAdmissionRejection(409, { error: { code: 'admission_uncertain', message: 'manual review', traceId: 't' } })).toBe(false); // A flat top-level {code} is not the canonical envelope — must not match. expect(isTransientAdmissionRejection(503, { code: 'delivery_busy' })).toBe(false); }); + it('keeps admission_uncertain separate from transient rejections', () => { + const uncertainEnvelope = { error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }; + expect(isTransientAdmissionRejection(409, uncertainEnvelope)).toBe(false); + expect(isTransientAdmissionRejection(409, JSON.stringify(uncertainEnvelope))).toBe(false); + expect(isAdmissionUncertain(409, uncertainEnvelope)).toBe(true); + expect(isAdmissionUncertain(409, JSON.stringify(uncertainEnvelope))).toBe(true); + expect(isAdmissionUncertain(503, uncertainEnvelope)).toBe(false); + expect(isAdmissionUncertain(409, { error: { code: 'internal', message: 'x' } })).toBe(false); + expect(isAdmissionUncertain(409, { code: 'admission_uncertain' })).toBe(false); + }); + it('isTerminalBridgeTask detects done/failed only', () => { expect(isTerminalBridgeTask(makeTask({ status: 'done' }))).toBe(true); expect(isTerminalBridgeTask(makeTask({ status: 'failed' }))).toBe(true); diff --git a/app/desktop/src/hooks/hubIntegrationMappers.ts b/app/desktop/src/hooks/hubIntegrationMappers.ts index ec3b4d8de..af2d1d918 100644 --- a/app/desktop/src/hooks/hubIntegrationMappers.ts +++ b/app/desktop/src/hooks/hubIntegrationMappers.ts @@ -339,25 +339,48 @@ export function extractCreatedRunId(value: unknown): string { } /** - * Classify a definite transient Edge admission rejection. Only these two codes - * are safe to leave queued for the Hub outbox to retry: a busy delivery slot - * (503 delivery_busy) and a Hub thread already occupied by an active run - * (409 active_run_exists). Any other non-OK response keeps the existing - * failure handling rather than being treated as a retryable admission. + * Read the canonical Edge error code from `{ error: { code, message, traceId } }`. + * Accepts an already-parsed object or a raw JSON string. + */ +export function readEdgeErrorCode(body: unknown): string | undefined { + const record = parseRecord(body); + return getFirstString(parseRecord(record.error).code); +} + +/** + * Classify admission outcomes where the Edge cannot safely confirm this + * delivery was accepted, so the Hub outbox must retain ownership for a retry. + * + * Safe to leave queued without ACK/FAIL: + * - 503 delivery_busy: pending admission contention or capacity; includes Retry-After. + * - 409 active_run_exists: another active run owns the thread; not this delivery. + * - 429 too_many_concurrent_runs: executor capacity rejected before accepting; same ID may retry. + * - 503 admission_persist_failed: admission evidence was not persisted. The executor may + * already have accepted or completed work, so do not ACK/FAIL; a replay only + * re-checks Edge evidence and does not assert that execution never happened. * - * Reads the canonical Edge error envelope ({ error: { code, message, traceId } }), - * which may arrive either as an already-parsed object or as a raw JSON string. + * `409 admission_uncertain` is deliberately NOT in this set: its outcome is + * unknown and requires manual review, so it is neither temporary nor accepted. */ export function isTransientAdmissionRejection(status: number, body: unknown): boolean { - const record = parseRecord(body); - const error = parseRecord(record.error); - const code = getFirstString(error.code); + const code = readEdgeErrorCode(body); return ( (status === 503 && code === 'delivery_busy') || - (status === 409 && code === 'active_run_exists') + (status === 409 && code === 'active_run_exists') || + (status === 429 && code === 'too_many_concurrent_runs') || + (status === 503 && code === 'admission_persist_failed') ); } +/** + * Detect `409 admission_uncertain`, a distinct non-final admission outcome. + * The caller must keep the task queued with a manual-review error and must not + * ACK/FAIL/relay-ACK or treat it as a retryable rejection or an accepted run. + */ +export function isAdmissionUncertain(status: number, body: unknown): boolean { + return status === 409 && readEdgeErrorCode(body) === 'admission_uncertain'; +} + export const FINAL_OUTPUT_MAX_CHARS = 32_000; /** diff --git a/app/desktop/src/hooks/useHubIntegration.ts b/app/desktop/src/hooks/useHubIntegration.ts index 0666a3e4b..3f59ce657 100644 --- a/app/desktop/src/hooks/useHubIntegration.ts +++ b/app/desktop/src/hooks/useHubIntegration.ts @@ -35,6 +35,7 @@ import { getTeamRouteContext, hasTaskProgressed, isTerminalBridgeTask, + isAdmissionUncertain, isTransientAdmissionRejection, normalizeRuntimeAgentId, parsePermissionDecisionControl, @@ -401,8 +402,10 @@ export function useHubIntegration(options: HubIntegrationOptions): HubIntegratio ); if (!runResp.ok) { - // Classify definite transient admission rejections (delivery_busy / - // active_run_exists). Anything else keeps the existing permanent path. + // Classify definite transient admission rejections (delivery_busy, + // active_run_exists, too_many_concurrent_runs, admission_persist_failed). + // Anything else keeps the existing permanent path unless it is an + // explicit admission_uncertain, which needs manual review. const errorText = await runResp.text().catch(() => 'Unknown error'); if (isTransientAdmissionRejection(runResp.status, errorText)) { // Keep the task queued/waiting; retry ownership stays with the Hub @@ -410,7 +413,19 @@ export function useHubIntegration(options: HubIntegrationOptions): HubIntegratio // invent a second retry loop here. return; } - throw new Error(`Edge POST /v1/runs returned ${runResp.status}: ${errorText}`); + if (isAdmissionUncertain(runResp.status, errorText)) { + // The executor did not return a definite outcome. Preserve the + // queued task and Edge-provided message for manual review; do not + // ACK/FAIL/relay-ACK and do not start a client-side retry. + const currentTask = store.getState().tasks.find((t) => t.taskId === taskId); + if (!hasTaskProgressed(currentTask)) { + store.getState().updateTask(taskId, { + error: 'Edge admission result is uncertain (HTTP ' + runResp.status + '): ' + errorText, + }); + } + return; + } + throw new Error('Edge POST /v1/runs returned ' + runResp.status + ': ' + errorText); } const runId = extractCreatedRunId(await runResp.json()); @@ -422,8 +437,10 @@ export function useHubIntegration(options: HubIntegrationOptions): HubIntegratio const taskProgressed = hasTaskProgressed(existingTask); // Business state/mapping is only written once (first accepted instance). + // Also clear a stale admission_uncertain manual-review note. Existing + // progressed state keeps its own error untouched. if (!taskProgressed) { - store.getState().updateTask(taskId, { runId, status: 'running' }); + store.getState().updateTask(taskId, { runId, status: 'running', error: undefined }); } // Every accepted delivery (first accept AND successful replay) is diff --git a/app/desktop/src/stores/taskBridgeStore.ts b/app/desktop/src/stores/taskBridgeStore.ts index 82be8b8f6..6f7286976 100644 --- a/app/desktop/src/stores/taskBridgeStore.ts +++ b/app/desktop/src/stores/taskBridgeStore.ts @@ -12,7 +12,8 @@ export interface AgentTask { runId?: string; status: 'queued' | 'running' | 'done' | 'failed'; dispatchPayload: Record; - error?: string; + /** Explicit undefined clears a previous error through updateTask. */ + error?: string | undefined; /** Timestamp when the dispatch was received. */ createdAt: string; } From 4cc6b67ac931c26d9920af9c4a2fb74006b00bb8 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Mon, 7 Sep 2026 05:49:24 +0800 Subject: [PATCH 3/5] docs(api): separate admission receipts from execution recovery Describe persistent admission evidence, cached and cold replay authorization, retryable pre-execution rejection and explicit uncertain outcomes. Expose admission metadata without claiming process recovery. Refs #2349. Co-authored-by: Codex --- api/events.md | 6 +++++- api/openapi.yaml | 28 +++++++++++++++++++++++++--- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/api/events.md b/api/events.md index 4f7741437..e6b5524b9 100644 --- a/api/events.md +++ b/api/events.md @@ -40,7 +40,11 @@ Hub 的 WS `agent.dispatch` 与 outbox HTTP POST `/v1/runs` 共享同一 `delive - **成功重放**：有效回执返回原 run 的正常 202 envelope，包括 `data.runId`、`data.deduplicated: true` 和 `data.deliveryId`；不新建 run、timeline 或 executor。原 run 已删除则返回 404 `not_found`，不静默重建。每次请求先通过 capability 校验；重放还按原 run 实际 project/thread scope 复验。 - **临时拒绝**：同 ID 正在接收，或容量被 pending claim 占满时，返回 503 `delivery_busy` + `Retry-After`（秒），而不是成功回执。409 `active_run_exists` 表示线程被其他活动 run 占用，也不能当作本次投递成功。Desktop 对这两类拒绝不 ACK、不 FAIL，由现有 Hub outbox 负责重投。 - **ACK 与业务状态**：每次成功接收或重放都幂等重发 task / relay ACK，以修复丢失的确认；已建立的 run 映射、输出和 running/terminal 状态不因重复投递而回退，业务接收通知只触发一次。 -- **进程边界**：回执不是持久化执行日志，重启后会丢失；跨重启身份核对与执行恢复不由此缓存保证。`queued` 状态本身不能证明子进程尚未启动，不能据此自动重启旧 run。 +- **Hub task 接收证据**：非空 `hubTaskId` 在启动执行器前，与 `admissionState: pending` 一起写入 run；File/SQLite 在返回前同步保存。执行器返回后只允许转为 `accepted` 或带 `admissionErrorCode` 的 `rejected`，不改写执行状态。没有 Hub task 的本地/MCP 请求保留原路径。 +- **冷重放**：进程缓存丢失或 delivery ID 改变时，按 Hub task 查最新 attempt，并复验原 run scope。`accepted` 返回原 run，不重新执行；上次最终证据保存失败时只重试保存。已接收 run 后续 `failed` 不等于接收拒绝。 +- **拒绝与未决**：429 `too_many_concurrent_runs` 是执行器持有执行权之前的容量拒绝；503 `admission_persist_failed` 是证据保存失败（执行器可能已经接收），两者都不应 ACK/FAIL，由 Hub 重投向 Edge 核对。只有明确的容量拒绝或调用 Start 之前的保存失败，才允许创建新 attempt；普通 `executor_start_failed` 保持拒绝，不伪装成功。 +- **结果不明**：同一 Hub task 的当前接收者仍在处理时返回 503 `delivery_busy`。恢复后的 `pending`、未知 admission state、无 `startedAt` 的旧 run 返回 409 `admission_uncertain`；Desktop 保持待核对错误，不 ACK/FAIL、不自动启动。旧 run 只有明确 `startedAt` 才能按原身份重放。`GET /v1/runs/{runId}` 暴露已记录的 `admissionState` / `admissionErrorCode` 供核对。 +- **恢复边界**：缓存不是恢复日志；持久化接收证据证明的是是否接收，不保证进程仍在运行，也不提供自动进程恢复。`queued`/`failed` 本身不能证明没有外部副作用。未决 admission 不参与终态自动清理；原 run 因显式删除或正常 retention 消失后，不宣称永久保留 Hub task 的幂等身份。 标签：**UPSERT by id**（稳定 id 合并，禁止第二行）；**idempotent on apply**（再应用不变）；**水位 / watermark**（只前进 `max`）；**ephemeral**（可丢可重，不写持久态）；**非幂等**（须自备去重或 REST）。 diff --git a/api/openapi.yaml b/api/openapi.yaml index f93e0cb22..84d09d82e 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -729,8 +729,21 @@ paths: "409": # delivery_conflict: delivery_id binds a different Hub task or legacy # project/thread scope (#2347); active_run_exists: the thread already has - # an active run. Client must not treat it as the same work. + # an active run. admission_uncertain requires reconciliation of a retained + # run without proof of admission; clients must not ACK/FAIL or restart. $ref: "#/components/responses/Error" + "429": + description: Executor capacity rejected admission before execution; retry the same Hub task later. + headers: + Retry-After: + description: Delay in seconds before retrying admission. + schema: + type: integer + minimum: 1 + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" "404": # not_found: the original run referenced by an accepted receipt was removed. $ref: "#/components/responses/Error" @@ -739,10 +752,12 @@ paths: not_configured means the Edge cannot validate a Hub identity because its Hub credential policy is unconfigured; it fails closed. delivery_busy is temporary contention for a pending delivery or - admission capacity and includes Retry-After. Neither is acceptance. + admission capacity and includes Retry-After. admission_persist_failed + means admission evidence could not be saved; execution may already + have been accepted. Retry to reconcile, not to force another start. headers: Retry-After: - description: Delay in seconds, present for delivery_busy responses. + description: Delay in seconds, present for delivery_busy and admission_persist_failed responses. schema: type: integer minimum: 1 @@ -9593,6 +9608,13 @@ components: finishedAt: type: string format: date-time + admissionState: + type: string + enum: [pending, accepted, rejected] + description: Hub task admission evidence, independent of execution status. Absent on legacy or non-Hub runs. + admissionErrorCode: + type: string + description: Public-safe rejection code recorded with rejected admission; not an execution failure reason. deduplicated: type: boolean description: True on an accepted replay of an already admitted delivery (same runId returned, no new run created). From 96c3ec04c5f53e9e2fb20fa473f95d11eb151d74 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Mon, 7 Sep 2026 05:57:04 +0800 Subject: [PATCH 4/5] refactor(dispatch): separate admission decisions from durable record writes Keep replay/target decisions distinct from the pending identity write so the state machine remains within the existing complexity gate. Serialize persistence failure injection with the background writer. Co-authored-by: Codex --- edge-server/internal/runcontrol/admission.go | 25 +++++++++++++------ .../store/store_run_admission_test.go | 4 +++ 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/edge-server/internal/runcontrol/admission.go b/edge-server/internal/runcontrol/admission.go index 5b2213579..252839b9a 100644 --- a/edge-server/internal/runcontrol/admission.go +++ b/edge-server/internal/runcontrol/admission.go @@ -61,6 +61,20 @@ func prepareRunAdmission(repository store.Repository, executor lifecycle.RunExec if params.HubTaskID != "" && params.BuildContext == nil { return store.Run{}, false, errcode.ErrExecutorUnavailable.WithMessage("Hub task admission requires an executor context") } + run, err := createRunAdmissionRecord(repository, params) + if err != nil { + return store.Run{}, false, err + } + if params.HubTaskID != "" { + pendingRunAdmissions[run.ID] = struct{}{} + } + return run, false, nil +} + +// createRunAdmissionRecord persists pending Hub identity before Start can run. +// A failed write retains a definite pre-execution rejection in this process; +// this is distinct from replay authorization and execution-state decisions. +func createRunAdmissionRecord(repository store.Repository, params CreateParams) (store.Run, *errcode.Error) { runID := generateRunID() var run store.Run var err error @@ -71,7 +85,7 @@ func prepareRunAdmission(repository store.Repository, executor lifecycle.RunExec } if err != nil { if errors.Is(err, store.ErrNotFound) { - return store.Run{}, false, errcode.ErrNotFound.WithMessage("project or thread not found") + return store.Run{}, errcode.ErrNotFound.WithMessage("project or thread not found") } if params.HubTaskID != "" { // This attempt has not called Start. Keep that definite rejection in @@ -81,14 +95,11 @@ func prepareRunAdmission(repository store.Repository, executor lifecycle.RunExec _, _ = repository.RecordRunAdmission(runID, errcode.ErrAdmissionPersistFailed.Code) repository.SetRunStatusIf(runID, "failed", "queued") } - return store.Run{}, false, errcode.ErrAdmissionPersistFailed + return store.Run{}, errcode.ErrAdmissionPersistFailed } - return store.Run{}, false, errcode.ErrInternal.WithMessage("failed to create run") - } - if params.HubTaskID != "" { - pendingRunAdmissions[run.ID] = struct{}{} + return store.Run{}, errcode.ErrInternal.WithMessage("failed to create run") } - return run, false, nil + return run, nil } func authorizeRunReplay(run store.Run, params CreateParams) *errcode.Error { diff --git a/edge-server/internal/store/store_run_admission_test.go b/edge-server/internal/store/store_run_admission_test.go index d431fa252..b63455bd2 100644 --- a/edge-server/internal/store/store_run_admission_test.go +++ b/edge-server/internal/store/store_run_admission_test.go @@ -434,6 +434,8 @@ func blockRunAdmissionPersistence(t *testing.T, repo Repository) (func(), func() switch r := repo.(type) { case *FileStore: return func() { + r.persistMu.Lock() + defer r.persistMu.Unlock() if err := os.Remove(r.path); err != nil && !errors.Is(err, os.ErrNotExist) { t.Fatalf("remove file snapshot before block: %v", err) } @@ -441,6 +443,8 @@ func blockRunAdmissionPersistence(t *testing.T, repo Repository) (func(), func() t.Fatalf("mkdir snapshot path to block rename: %v", err) } }, func() { + r.persistMu.Lock() + defer r.persistMu.Unlock() if err := os.Remove(r.path); err != nil { t.Fatalf("remove snapshot path block: %v", err) } From a66dc2e2e2f3b22fcdd351699b4be8b805bb14f1 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Mon, 7 Sep 2026 06:05:12 +0800 Subject: [PATCH 5/5] fix(desktop): surface uncertain admission for user review Do not leave the manual-review reason in a hidden bridge store. Show it through the existing warning UI and deduplicate unchanged reasons across transport trace IDs; keep acceptance and execution unchanged. Co-authored-by: Codex --- api/events.md | 2 +- .../__e2e__/hub-delivery-admission.spec.ts | 4 ++ .../src/__tests__/useHubIntegration.test.ts | 71 +++++++++++-------- app/desktop/src/hooks/useHubIntegration.ts | 12 +++- 4 files changed, 55 insertions(+), 34 deletions(-) diff --git a/api/events.md b/api/events.md index e6b5524b9..ac3f7fcce 100644 --- a/api/events.md +++ b/api/events.md @@ -43,7 +43,7 @@ Hub 的 WS `agent.dispatch` 与 outbox HTTP POST `/v1/runs` 共享同一 `delive - **Hub task 接收证据**：非空 `hubTaskId` 在启动执行器前，与 `admissionState: pending` 一起写入 run；File/SQLite 在返回前同步保存。执行器返回后只允许转为 `accepted` 或带 `admissionErrorCode` 的 `rejected`，不改写执行状态。没有 Hub task 的本地/MCP 请求保留原路径。 - **冷重放**：进程缓存丢失或 delivery ID 改变时，按 Hub task 查最新 attempt，并复验原 run scope。`accepted` 返回原 run，不重新执行；上次最终证据保存失败时只重试保存。已接收 run 后续 `failed` 不等于接收拒绝。 - **拒绝与未决**：429 `too_many_concurrent_runs` 是执行器持有执行权之前的容量拒绝；503 `admission_persist_failed` 是证据保存失败（执行器可能已经接收），两者都不应 ACK/FAIL，由 Hub 重投向 Edge 核对。只有明确的容量拒绝或调用 Start 之前的保存失败，才允许创建新 attempt；普通 `executor_start_failed` 保持拒绝，不伪装成功。 -- **结果不明**：同一 Hub task 的当前接收者仍在处理时返回 503 `delivery_busy`。恢复后的 `pending`、未知 admission state、无 `startedAt` 的旧 run 返回 409 `admission_uncertain`；Desktop 保持待核对错误，不 ACK/FAIL、不自动启动。旧 run 只有明确 `startedAt` 才能按原身份重放。`GET /v1/runs/{runId}` 暴露已记录的 `admissionState` / `admissionErrorCode` 供核对。 +- **结果不明**：同一 Hub task 的当前接收者仍在处理时返回 503 `delivery_busy`。恢复后的 `pending`、未知 admission state、无 `startedAt` 的旧 run 返回 409 `admission_uncertain`；Desktop 保持待核对错误并通过现有通知提示用户，同一原因不重复提示，不 ACK/FAIL、不自动启动。旧 run 只有明确 `startedAt` 才能按原身份重放。`GET /v1/runs/{runId}` 暴露已记录的 `admissionState` / `admissionErrorCode` 供核对。 - **恢复边界**：缓存不是恢复日志；持久化接收证据证明的是是否接收，不保证进程仍在运行，也不提供自动进程恢复。`queued`/`failed` 本身不能证明没有外部副作用。未决 admission 不参与终态自动清理；原 run 因显式删除或正常 retention 消失后，不宣称永久保留 Hub task 的幂等身份。 标签：**UPSERT by id**（稳定 id 合并，禁止第二行）；**idempotent on apply**（再应用不变）；**水位 / watermark**（只前进 `max`）；**ephemeral**（可丢可重，不写持久态）；**非幂等**（须自备去重或 REST）。 diff --git a/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts b/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts index e902ac45c..0b615a6b0 100644 --- a/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts +++ b/app/desktop/src/__e2e__/hub-delivery-admission.spec.ts @@ -233,6 +233,10 @@ for (const theme of ['light', 'dark'] as const) { await expect.poll(() => calls.runs.length).toBe(4); await expect.poll(() => readTaskState(page)).toEqual(queued); await expect.poll(() => readTaskError(page)).toContain('fixture admission rejection: admission_uncertain'); + const reviewNotice = page.getByText(/Edge admission result is uncertain.*fixture admission rejection: admission_uncertain/); + await expect(reviewNotice).toBeVisible(); + await expect(reviewNotice).toHaveCount(1); + await page.screenshot({ path: testInfo.outputPath('admission-uncertain-' + theme + '.png') }); expect(calls.acks).toHaveLength(0); expect(calls.relayAcks).toHaveLength(0); expect(calls.fails).toHaveLength(0); diff --git a/app/desktop/src/__tests__/useHubIntegration.test.ts b/app/desktop/src/__tests__/useHubIntegration.test.ts index dc5d18b5f..53e25198a 100644 --- a/app/desktop/src/__tests__/useHubIntegration.test.ts +++ b/app/desktop/src/__tests__/useHubIntegration.test.ts @@ -118,6 +118,7 @@ import { createServer, type IncomingMessage } from 'node:http'; import type { HubWSHandle } from '@shared/hub/hubWS'; import type { HubClient } from '@/api/hubClient'; import { HUB_EVENTS } from '@shared/hubEvents'; +import { useToastStore } from '@shared/ui/toast'; import { useHubIntegration } from '@/hooks/useHubIntegration'; // ── Helpers ───────────────────────────────────────────── @@ -776,38 +777,48 @@ describe('useHubIntegration', () => { }); it('keeps an admission_uncertain rejection queued for manual review without acking or failing', async () => { - mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }, 409); - renderHook(() => - useHubIntegration({ - hubWS, - hubClient, - dispatchTarget: { targetId: 'target-current', deviceId: 'desktop-current' }, - }), - ); - - const relayFrame = { - relay_command_id: 'relay-1', - command_type: 'agent.dispatch', - payload: JSON.stringify( - makeDispatchPayload({ - target_id: 'target-current', - edge_device_id: 'desktop-current', - delivery_id: 'd1', + const notice = vi.spyOn(useToastStore.getState(), 'showToast').mockReturnValue('notice-fixture'); + try { + mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_001' } }, 409); + renderHook(() => + useHubIntegration({ + hubWS, + hubClient, + dispatchTarget: { targetId: 'target-current', deviceId: 'desktop-current' }, }), - ), - }; - await act(async () => { - fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); - }); + ); - expect(hoisted.storeTasks).toHaveLength(1); - expect(hoisted.storeTasks[0]?.status).toBe('queued'); - expect(hoisted.storeTasks[0]?.runId).toBeUndefined(); - expect(hoisted.storeTasks[0]?.error).toContain('Edge admission result is uncertain'); - expect(hoisted.storeTasks[0]?.error).toContain('manual review'); - expect(hubClient.ackTask).not.toHaveBeenCalled(); - expect(hubClient.failTask).not.toHaveBeenCalled(); - expect(hubClient.ackRelayCommand).not.toHaveBeenCalled(); + const relayFrame = { + relay_command_id: 'relay-1', + command_type: 'agent.dispatch', + payload: JSON.stringify( + makeDispatchPayload({ + target_id: 'target-current', + edge_device_id: 'desktop-current', + delivery_id: 'd1', + }), + ), + }; + await act(async () => { + fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); + }); + + expect(hoisted.storeTasks).toHaveLength(1); + expect(hoisted.storeTasks[0]?.status).toBe('queued'); + expect(hoisted.storeTasks[0]?.runId).toBeUndefined(); + expect(hoisted.storeTasks[0]?.error).toContain('Edge admission result is uncertain'); + expect(hoisted.storeTasks[0]?.error).toContain('manual review'); + expect(hubClient.ackTask).not.toHaveBeenCalled(); + expect(hubClient.failTask).not.toHaveBeenCalled(); + expect(hubClient.ackRelayCommand).not.toHaveBeenCalled(); + + mockRunCreateResponseWithStatus({ error: { code: 'admission_uncertain', message: 'manual review', traceId: 'trace_002' } }, 409); + await act(async () => { fireHubEvent(HUB_EVENTS.AGENT_DISPATCH, relayFrame); }); + expect(notice).toHaveBeenCalledTimes(1); + expect(notice).toHaveBeenCalledWith('warning', expect.stringContaining('manual review'), { duration: 10_000 }); + } finally { + notice.mockRestore(); + } }); it('clears a queued admission_uncertain error when the same delivery is accepted', async () => { diff --git a/app/desktop/src/hooks/useHubIntegration.ts b/app/desktop/src/hooks/useHubIntegration.ts index 3f59ce657..e229902b4 100644 --- a/app/desktop/src/hooks/useHubIntegration.ts +++ b/app/desktop/src/hooks/useHubIntegration.ts @@ -20,6 +20,7 @@ import type { CoordinatorRouteDecision, HubClient } from '@/api/hubClient'; import { createEventStream, type StreamHandle } from '@/api/eventClient'; import type { EventEnvelope } from '@shared/events'; import { HUB_EVENTS } from '@shared/hubEvents'; +import { useToastStore } from '@shared/ui/toast'; import { hubQueryKeys } from '@shared/stores/queryKeys'; import { useTaskBridgeStore, type AgentTask } from '@/stores/taskBridgeStore'; import { queryClient } from '@/api/queryClient'; @@ -419,9 +420,14 @@ export function useHubIntegration(options: HubIntegrationOptions): HubIntegratio // ACK/FAIL/relay-ACK and do not start a client-side retry. const currentTask = store.getState().tasks.find((t) => t.taskId === taskId); if (!hasTaskProgressed(currentTask)) { - store.getState().updateTask(taskId, { - error: 'Edge admission result is uncertain (HTTP ' + runResp.status + '): ' + errorText, - }); + const message = getString(parseRecord(parseRecord(errorText).error), 'message') || errorText; + const error = 'Edge admission result is uncertain (HTTP ' + runResp.status + '): ' + message; + store.getState().updateTask(taskId, { error }); + // The bridge has no task-error panel. Surface the need for review + // through the existing notification UI, once per unchanged reason. + if (currentTask?.error !== error) { + useToastStore.getState().showToast('warning', error, { duration: 10_000 }); + } } return; }