diff --git a/agent/worker.go b/agent/worker.go index 2c77722..84f693b 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strconv" + "strings" "time" "github.com/Autumn-27/artex/db" @@ -68,6 +69,30 @@ type Worker struct { nonStreamingFn func() bool } +// WorkerSessionID returns the stable transcript key used by a worker intent. +// Worker slots are reusable, so the intent id (rather than work#N) is the +// session identity. Keep this helper public so the Worker message API and UI +// can refer to exactly the conversation that will be resumed. +func WorkerSessionID(explorationID, intentID int64) string { + return fmt.Sprintf("exp%d-worker-i%d", explorationID, intentID) +} + +const workerChatMarkerPrefix = "" +} + +func hasWorkerChatMessage(messages []llm.Message, requestID string) bool { + marker := workerChatMarker(requestID) + for _, message := range messages { + if message.Role == llm.RoleUser && strings.Contains(message.Text(), marker) { + return true + } + } + return false +} + // SetNonStreaming wires a resolver deciding whether runs use the non-streaming // model path (true = non-streaming). nil/unset = streaming (default). Read per // run so a profile or task-chain toggle takes effect without rebuilding. @@ -274,6 +299,18 @@ func renderWorkerGraphOverview(data map[string]any) string { // explored but persisted nothing isn't mistaken for done, and the engine can log // facts/assets/findings separately instead of lumping them under "facts"). func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger, notifyFinding func(int64, string)) (harness.TerminalReason, WriteCounts, error) { + return w.execute(ctx, name, taskID, as, ts, intent, hooks, emit, enr, notifyFinding, "", "") +} + +// ExecuteWithMessage runs the next turn in the same intent conversation with a +// human-authored message. The HTTP handler does not edit the transcript; +// agentcore records the message as a normal user turn when this Worker starts. +// This keeps Worker continuation identical to the regular agent chat flow. +func (w *Worker) ExecuteWithMessage(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger, notifyFinding func(int64, string), requestID, message string) (harness.TerminalReason, WriteCounts, error) { + return w.execute(ctx, name, taskID, as, ts, intent, hooks, emit, enr, notifyFinding, strings.TrimSpace(requestID), strings.TrimSpace(message)) +} + +func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger, notifyFinding func(int64, string), requestID, message string) (harness.TerminalReason, WriteCounts, error) { tsx := NewToolSet(ts, name) tsx.SetTaskID(taskID) coverageEnabled := as == nil || as.CoverageEnabled(taskID) @@ -358,7 +395,7 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db. } if w.tx != nil { // persist raw LLM conversation; one file per worked intent opts.Transcript = w.tx - opts.SessionID = fmt.Sprintf("exp%d-worker-i%d", ts.ID(), intent.ID) + opts.SessionID = WorkerSessionID(ts.ID(), intent.ID) } intentID := intent.ID emitWrap := func(r db.Activity) { @@ -398,11 +435,26 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db. // being re-run. The transcript ID is deterministic per intent, so if a prior // session exists the worker continues from where it left off instead of // restarting from scratch. + alreadyRecorded := false if w.tx != nil { _ = s.Resume(opts.SessionID) - if len(s.Messages()) > 0 { + alreadyRecorded = requestID != "" && hasWorkerChatMessage(s.Messages(), requestID) + if len(s.Messages()) > 0 && message == "" { seedUnlockFromHistory(s.Messages(), def.UnlockSkill) input = "继续执行。" + } else if len(s.Messages()) > 0 { + seedUnlockFromHistory(s.Messages(), def.UnlockSkill) + } + } + if message != "" { + if alreadyRecorded { + input = "继续执行上一次人工对话输入的新意图。不要重复已经完成的动作。" + } else if len(s.Messages()) > 0 { + input = workerChatMarker(requestID) + "\n【人工对话输入的新意图】\n" + message + + "\n\n请立即按这条人工输入执行，完成后再根据上下文决定原任务是否需要继续。" + } else { + input += "\n\n" + workerChatMarker(requestID) + "\n【人工对话输入的新意图】\n" + message + + "\n\n请优先执行这条人工输入。" } } diff --git a/agent/worker_intervention_test.go b/agent/worker_intervention_test.go new file mode 100644 index 0000000..ab8977e --- /dev/null +++ b/agent/worker_intervention_test.go @@ -0,0 +1,28 @@ +package agent + +import ( + "testing" + + "github.com/Autumn-27/norma/llm" +) + +func TestWorkerSessionIDIsStablePerIntent(t *testing.T) { + if got := WorkerSessionID(12, 34); got != "exp12-worker-i34" { + t.Fatalf("session id = %q, want exp12-worker-i34", got) + } +} + +func TestWorkerChatMarkerOnlyMatchesItsNormalUserTurn(t *testing.T) { + requestID := "worker-message-123" + messages := []llm.Message{ + {Role: llm.RoleAssistant, Content: []llm.ContentBlock{llm.TextBlock(workerChatMarker(requestID))}}, + llm.UserText(workerChatMarker("worker-message-other") + "\nother"), + llm.UserText(workerChatMarker(requestID) + "\nnew intent"), + } + if !hasWorkerChatMessage(messages, requestID) { + t.Fatal("expected the matching user turn to be detected") + } + if hasWorkerChatMessage(messages, "worker-message-missing") { + t.Fatal("unrelated request id matched a Worker user turn") + } +} diff --git a/server/engine.go b/server/engine.go index f3c7074..b0b0a56 100644 --- a/server/engine.go +++ b/server/engine.go @@ -913,15 +913,28 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { } } -// runWorkerStep claims and fully settles at most one intent. Its caller holds one -// task-operation admission for the whole claim/execute/state-write sequence so a -// delete cannot observe quiescence between the LLM return and the final DB writes. +// runWorkerStep claims one intent from the frontier and fully settles it via +// runIntent. Returns false when nothing was claimable. The pool worker loop is its +// only caller. func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker *agent.Worker) bool { intent := e.claimNext(t, name) if intent == nil { return false } log.Printf("[worker %s] task %s 领取意图 #%d", name, t.ID, intent.ID) + return e.runIntent(ctx, t, name, worker, intent, "", "") +} + +// runIntent executes and fully settles one already-claimed (state=running) intent. +// Both the pool worker loop (via runWorkerStep) and the human-message handler (via +// runDetachedIntent, a dedicated goroutine outside the worker pool) call it, so the +// execute/retry/state-write logic lives in exactly one place. A non-empty message +// is injected as this turn's input through ExecuteWithMessage; requestID keys the +// transcript marker that dedups re-injection across model_error retries. The caller +// must already hold one task-operation admission for the whole sequence so a delete +// cannot observe quiescence between the LLM return and the final DB writes. +func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *agent.Worker, intent *db.Node, requestID, message string) bool { + hasChatMessage := message != "" e.stampFirstRun(t) // 首次真正执行 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务) e.touch(t.ID) emit := func(r db.Activity) { e.emitActivity(t, r) } @@ -947,7 +960,14 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker hooks := steerHooks{inner: t.Guard.Hooks(), drain: func() (string, bool) { return e.drainSteer(iid) }} wTaskID, _ := strconv.ParseInt(t.ID, 10, 64) e.BeginLLMCall(t.ID) - reason, wrote, err := worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding) + var reason harness.TerminalReason + var wrote agent.WriteCounts + var err error + if hasChatMessage { + reason, wrote, err = worker.ExecuteWithMessage(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding, requestID, message) + } else { + reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding) + } e.EndLLMCall(t.ID) // model_error 收场 → 额外重跑几次（退避后再试）。仅在意图仍属本 work、任务 // 未暂停/未终止/未取消【且未进入收尾】时重试；否则让位给对应分支处理(收尾期不 @@ -961,7 +981,11 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker break // 退避期间被取消（终止/暂停）→ 交给下方分支处理 } e.BeginLLMCall(t.ID) - reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding) + if hasChatMessage { + reason, wrote, err = worker.ExecuteWithMessage(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding, requestID, message) + } else { + reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding) + } e.EndLLMCall(t.ID) } // Capture kill state before detachWork cancels workCtx. kill = this work's @@ -1076,6 +1100,61 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker return true } +// runDetachedIntent runs one paused intent OUTSIDE the worker pool in its own +// goroutine — the human-message path. It transitions the intent paused->running +// itself (never through 'open'), so the pool, which only claims 'open', can never +// race it; the "at most one run per intent" invariant still holds because winning +// the CAS is the sole entry and work[intentID] was cleared when the pause settled. +// Because it does not compete for a frontier slot, a user message continues the +// worker immediately even when all pool slots are busy (mirroring how the +// main-agent chat handler starts its run directly). The spawned goroutine owns one +// task-operation admission for the whole run and roots its context at ctx (pass the +// server root, never the HTTP request, so a disconnect cannot strand the run while +// task pause/delete/shutdown still stops it). Returns an error if the run could not +// be started; the intent is left untouched in that case. +func (e *Engine) runDetachedIntent(ctx context.Context, t *Task, intentID int64, requestID, message string) error { + if !e.beginTaskOperation(t.ID) { + return fmt.Errorf("task is being deleted") + } + release := true + defer func() { + if release { + e.decInflight(t.ID) + } + }() + _, worker := e.snapshotFor(t) + if worker == nil { + return fmt.Errorf("worker 尚未就绪") + } + node, err := t.Store.GetNode(intentID) + if err != nil { + return err + } + if node == nil || node.Kind != db.KindIntent { + return fmt.Errorf("intent not found") + } + changed, err := t.Store.CompareAndSetIntentState(intentID, "paused", "running") + if err != nil { + return err + } + if !changed { + return fmt.Errorf("%w: 意图不再是 paused 状态", db.ErrIntentStateConflict) + } + node.State, node.Owner = "running", "chat" + // Record the human turn as a visible activity BEFORE the run starts, so it is + // ordered ahead of any worker step and never appears without the run happening. + // This is the UI copy; ExecuteWithMessage separately writes it into the intent + // transcript as the LLM input. + uid := intentID + e.emitActivity(t, db.Activity{NodeID: &uid, Worker: "user", Kind: "user", Summary: message, Detail: message}) + release = false // ownership of the admission passes to the goroutine + go func() { + defer e.decInflight(t.ID) + e.runIntent(ctx, t, "chat", worker, node, requestID, message) + }() + return nil +} + func taskExecutionPaused(cause error) bool { var abort *agent.AbortCause if !errors.As(cause, &abort) { diff --git a/server/intent_intervention.go b/server/intent_intervention.go new file mode 100644 index 0000000..3e3655c --- /dev/null +++ b/server/intent_intervention.go @@ -0,0 +1,142 @@ +package server + +import ( + "encoding/json" + "errors" + "net/http" + "strconv" + "strings" + + "github.com/Autumn-27/artex/db" +) + +const maxWorkerMessageBytes = 64 << 10 + +func validWorkerMessageRequestID(id string) bool { + if id == "" || len(id) > 128 { + return false + } + for _, r := range id { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || + (r >= '0' && r <= '9') || r == '-' || r == '_' || r == '.' || r == ':' { + continue + } + return false + } + return true +} + +// sendWorkerMessage continues a paused Worker intent with a human-authored message. +// The message is injected as the next turn's input through the same +// resume-from-transcript path the worker uses on a normal resume (ExecuteWithMessage), +// so this reuses the existing pause/resume machinery rather than a bespoke protocol. +// The run happens in a dedicated goroutine outside the worker pool (runDetachedIntent), +// so the message is picked up immediately even when every pool slot is busy — the same +// way the main-agent chat handler starts its run directly. In-memory only: a process +// restart re-runs the intent from its transcript without the message, which is +// acceptable for this rare interrupt-then-continue action. +func (s *Server) sendWorkerMessage(w http.ResponseWriter, r *http.Request) { + t, ok := s.m.Task(r.PathValue("id")) + if !ok { + writeErr(w, http.StatusNotFound, "task not found") + return + } + iid, err := strconv.ParseInt(r.PathValue("iid"), 10, 64) + if err != nil || iid <= 0 { + writeErr(w, http.StatusBadRequest, "bad intent id") + return + } + r.Body = http.MaxBytesReader(w, r.Body, maxWorkerMessageBytes) + var req struct { + Message string `json:"message"` + RequestID string `json:"request_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + var tooLarge *http.MaxBytesError + if errors.As(err, &tooLarge) { + writeErr(w, http.StatusRequestEntityTooLarge, "请求体过大") + return + } + writeErr(w, http.StatusBadRequest, "bad json: "+err.Error()) + return + } + message := strings.TrimSpace(req.Message) + requestID := strings.TrimSpace(req.RequestID) + if message == "" { + writeErr(w, http.StatusBadRequest, "消息不能为空") + return + } + if len([]rune(message)) > 4000 { + writeErr(w, http.StatusBadRequest, "消息不能超过 4000 个字符") + return + } + if !validWorkerMessageRequestID(requestID) { + writeErr(w, http.StatusBadRequest, "request_id 必须是 1-128 位字母、数字、-、_、. 或 :") + return + } + + // Reject non-runnable task lifecycles up front so the caller gets a clear reason + // instead of a silent no-op. The intent itself must be paused: the UI flow is + // interrupt (pause) first, then send. + if s.engine.IsDeleting(t.ID) { + writeErr(w, http.StatusConflict, "任务正在删除，无法向 Worker 发送消息") + return + } + lifecycle := t.lifecycleSnapshot() + switch { + case lifecycle.Paused || s.engine.IsPaused(t.ID): + writeErr(w, http.StatusConflict, "任务已暂停，请先恢复任务再向 Worker 发送消息") + return + case lifecycle.Queued: + writeErr(w, http.StatusConflict, "排队中的任务无法向 Worker 发送消息") + return + case isTerminalStatus(lifecycle.Status): + writeErr(w, http.StatusConflict, "终态任务无法向 Worker 发送消息") + return + case s.engine.isSettling(t.ID): + writeErr(w, http.StatusConflict, "任务正在收尾，无法向 Worker 发送消息") + return + } + + node, err := t.Store.GetNode(iid) + if err != nil { + writeErr(w, http.StatusInternalServerError, err.Error()) + return + } + if node == nil { + if inherited, sourceErr := t.Store.GetNodeWithSources(iid); sourceErr == nil && inherited != nil && inherited.Inherited { + writeErr(w, http.StatusConflict, "继承意图为只读，不能发送 Worker 消息") + return + } + writeErr(w, http.StatusNotFound, "intent not found") + return + } + if node.Kind != db.KindIntent { + writeErr(w, http.StatusConflict, "node is not an intent") + return + } + if node.State != "paused" { + writeErr(w, http.StatusConflict, "仅已暂停的 Worker 可以发送消息，请先暂停") + return + } + + // runDetachedIntent transitions paused->running, emits the user turn and starts a + // dedicated run. Root the run at s.ctx so a disconnected browser cannot strand it + // while task pause/delete/shutdown still stop it. + if err := s.engine.runDetachedIntent(s.ctx, t, iid, requestID, message); err != nil { + switch { + case errors.Is(err, db.ErrIntentStateConflict): + writeErr(w, http.StatusConflict, err.Error()) + default: + writeErr(w, http.StatusInternalServerError, err.Error()) + } + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "id": iid, + "state": "running", + "accepted": true, + "request_id": requestID, + }) +} diff --git a/server/server.go b/server/server.go index f9ed5f4..f820b1e 100644 --- a/server/server.go +++ b/server/server.go @@ -749,6 +749,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("PUT /api/tasks/{id}/llm", s.updateTaskLLMProfiles) mux.HandleFunc("GET /api/tasks/{id}/llm/resolution", s.taskLLMResolutionHandler) mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/control", s.controlIntent) + mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/messages", s.sendWorkerMessage) mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/rerun", s.rerunIntent) // 重跑单条 blocked/exhausted/stopped 意图 mux.HandleFunc("POST /api/tasks/{id}/intents/rerun-blocked", s.rerunBlocked) // 批量重跑本任务全部 blocked 意图 mux.HandleFunc("POST /api/active", s.setActive) diff --git a/server/worker_message_test.go b/server/worker_message_test.go new file mode 100644 index 0000000..9487ba5 --- /dev/null +++ b/server/worker_message_test.go @@ -0,0 +1,54 @@ +package server + +import ( + "context" + "errors" + "strings" + "testing" +) + +// After the worker-message rework, the human-message path no longer goes through +// ControlWork with an "intervene" action — it runs the intent in a dedicated +// goroutine (runDetachedIntent) and pauses, if needed, with the ordinary "pause" +// action. Guard that ControlWork only knows pause|cancel so a reintroduced +// "intervene" caller fails loudly instead of silently no-op'ing. +func TestControlWorkRejectsRemovedInterveneAction(t *testing.T) { + e := NewEngine(nil) + err := e.ControlWork(context.Background(), 1, "intervene") + if err == nil || !containsUnsupported(err) { + t.Fatalf("intervene should be unsupported, got %v", err) + } +} + +// A control request for an intent with no live work is a conflict, not a panic or +// a silent success — the same guard the message handler relies on when a run has +// already finished between the UI read and the request. +func TestControlWorkConflictsWhenNoWorkRegistered(t *testing.T) { + e := NewEngine(nil) + err := e.ControlWork(context.Background(), 42, "pause") + if !errors.Is(err, errWorkControlConflict) { + t.Fatalf("pause with no work = %v, want errWorkControlConflict", err) + } +} + +// validWorkerMessageRequestID gates the idempotency token; keep its charset stable +// so a client-generated UUID is always accepted and injection-y ids are rejected. +func TestValidWorkerMessageRequestID(t *testing.T) { + ok := []string{"worker-message-abc123", "a.b:c_d-1", "550e8400-e29b-41d4-a716-446655440000"} + for _, id := range ok { + if !validWorkerMessageRequestID(id) { + t.Errorf("id %q should be valid", id) + } + } + bad := []string{"", "has space", "emoji😀", "slash/here", string(make([]byte, 129))} + for _, id := range bad { + if validWorkerMessageRequestID(id) { + t.Errorf("id %q should be rejected", id) + } + } +} + +func containsUnsupported(err error) bool { + return err != nil && !errors.Is(err, errWorkControlConflict) && + strings.Contains(err.Error(), "unsupported work action") +} diff --git a/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx b/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx index 937b97e..3ea1ffb 100644 --- a/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx +++ b/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx @@ -14,7 +14,6 @@ import { Loader2Icon, PaperclipIcon, PauseIcon, - PlayIcon, RadioIcon, RotateCwIcon, ShieldAlertIcon, @@ -43,6 +42,7 @@ import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { InputGroup, InputGroupAddon, InputGroupButton, InputGroupTextarea } from "@/components/ui/input-group"; import { ScrollArea } from "@/components/ui/scroll-area"; +import { Spinner } from "@/components/ui/spinner"; import { Textarea } from "@/components/ui/textarea"; import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "@/components/ui/tooltip"; import { api, sseUrl } from "@/lib/api"; @@ -93,6 +93,18 @@ const PAGE = 200; // history page size const SYSTEM_SCAN_PAGE = 500; // generic activity pages scanned to recover sparse system audit events const MAX_KEEP = 4000; // per-session in-memory cap; older pages re-fetched on scroll-up const STREAM_WINDOW_MS = 5000; // "live" = activity seen within this window +const MAX_WORKER_MESSAGE_CHARS = 4000; + +function newWorkerMessageRequestID(): string { + return ( + globalThis.crypto?.randomUUID?.() ?? + `worker-message-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 12)}` + ); +} + +function workerMessageCharCount(value: string): number { + return Array.from(value).length; +} // One session's lazily-loaded, reverse-paginated cache. `lastTs`/`unread` are kept // live even for sessions that were never opened, so the list shows liveness + unread @@ -313,8 +325,6 @@ function SessionItem({ hasPending, unread, onClick, - onPause, - onResume, onCancel, controlling, deleted, @@ -325,8 +335,6 @@ function SessionItem({ hasPending?: boolean; unread?: number; onClick: () => void; - onPause?: () => void; - onResume?: () => void; onCancel?: () => void; controlling?: boolean; deleted?: boolean; @@ -339,7 +347,8 @@ function SessionItem({ ) : null; - const controllable = s.role === "worker" && !s.inherited && (s.status === "running" || s.status === "paused"); + const cancellable = + s.role === "worker" && !s.inherited && !deleted && (s.status === "running" || s.status === "paused"); return (
)} - {controllable && ( + {cancellable && (
- {s.status === "running" ? ( - - ) : ( - - )}
+ ) : active.role === "worker" && + !active.inherited && + (active.status === "running" || active.status === "paused") ? ( +
+ + { + setWorkerMessage(event.target.value); + setWorkerMessageRequestId(""); + }} + onKeyDown={(event) => { + if (!shouldSubmitOnKey(event, sendMode)) return; + event.preventDefault(); + sendWorkerChat(); + }} + disabled={active.status === "running" || workerMessageSending} + aria-invalid={workerMessageCharCount(workerMessage) > MAX_WORKER_MESSAGE_CHARS} + className="max-h-36 min-h-9 overflow-y-auto" + /> + + MAX_WORKER_MESSAGE_CHARS && "text-destructive", + )} + > + {workerMessageCharCount(workerMessage)}/{MAX_WORKER_MESSAGE_CHARS} + + {active.status === "running" ? ( + void controlWorker(active, "pause")} + disabled={controllingIntent === active.intent_id} + title="暂停当前 Worker" + aria-label="暂停当前 Worker" + > + {controllingIntent === active.intent_id ? ( + + ) : ( + + )} + + ) : ( + <> + void controlWorker(active, "resume")} + disabled={controllingIntent === active.intent_id || workerMessageSending} + title="不发消息，直接继续执行" + aria-label="直接继续执行" + > + {controllingIntent === active.intent_id ? : "直接继续"} + + MAX_WORKER_MESSAGE_CHARS + } + title="发送消息" + aria-label="发送消息" + > + {workerMessageSending ? : } + + + )} + + +
) : (
(`/tasks/${taskId}/intents/${intentId}/control`, { action, reason: reason ?? "" }), + sendWorkerMessage: (taskId: string, intentId: string, message: string, requestId: string) => + post<{ + id: number; + state: "running"; + accepted: true; + request_id: string; + }>(`/tasks/${taskId}/intents/${intentId}/messages`, { message, request_id: requestId }), taskLLMResolution: (id: string) => get(`/tasks/${id}/llm/resolution`), // 重跑一条没跑成功的意图(blocked/exhausted/stopped)：置回 open，worker 会重新认领、从头再跑。 rerunIntent: (taskId: string, intentId: string) => diff --git a/web/src/lib/mock/handler.ts b/web/src/lib/mock/handler.ts index 407fe4e..efa66f5 100644 --- a/web/src/lib/mock/handler.ts +++ b/web/src/lib/mock/handler.ts @@ -10,6 +10,7 @@ import { normalizeCompanyScopeValue, } from "../company-scope"; import type { + Activity, ArchiveBatchItem, Asset, BatchControlItem, @@ -42,6 +43,7 @@ const mockConversations = structuredClone(D.conversations); const mockIntents = structuredClone(D.intents); const mockCompanies = structuredClone(D.companies); const mockAssets = structuredClone(D.assets); +const mockActivity = structuredClone(D.activity); type MockTaskArchiveSnapshot = { task: Task; numericTaskID: number; @@ -420,6 +422,15 @@ function bodyIDs(value: unknown): string[] { } type MockIntentControlResult = { ok: boolean; state?: string; error?: string }; +type MockWorkerMessage = { + intentId: string; + message: string; + state: "open" | "running"; + activitySeq: number; +}; + +const mockWorkerMessages = new Map(); +let nextMockWorkerMessageActivitySeq = mockActivity.reduce((maximum, item) => Math.max(maximum, item.seq), 0) + 1; function controlMockIntent(id: string, action: "pause" | "resume"): MockIntentControlResult { const intent = mockIntents.find((item) => item.id === id); @@ -432,6 +443,65 @@ function controlMockIntent(id: string, action: "pause" | "resume"): MockIntentCo return { ok: true, state: intent.state }; } +function sendMockWorkerMessage( + id: string, + message: string, + requestId: string, +): MockIntentControlResult & { activitySeq?: number; requestId?: string } { + const normalizedMessage = message.trim(); + const normalizedRequestId = requestId.trim(); + if (!normalizedRequestId) return { ok: false, error: "request_id 不能为空" }; + const previous = mockWorkerMessages.get(normalizedRequestId); + if (previous) { + if (previous.intentId !== id || previous.message !== normalizedMessage) { + return { ok: false, error: "request_id 已用于其他 Worker 消息" }; + } + return { + ok: true, + state: previous.state, + activitySeq: previous.activitySeq, + requestId: normalizedRequestId, + }; + } + const intent = mockIntents.find((item) => item.id === id); + if (!intent) return { ok: false, error: "意图不存在" }; + if (intent.inherited || intent.state !== "paused") { + return { ok: false, state: intent.state, error: "仅已暂停的 Worker 可以发送消息，请先暂停" }; + } + if (!normalizedMessage) return { ok: false, state: intent.state, error: "消息不能为空" }; + if (Array.from(normalizedMessage).length > 4000) { + return { ok: false, state: intent.state, error: "消息不能超过 4000 个字符" }; + } + + // The real endpoint transitions the intent paused->running, records the user turn, + // and runs it in a dedicated goroutine outside the worker pool. Keep the mock in + // sync: flip to running immediately and emit the visible user activity. + intent.state = "running"; + const activitySeq = nextMockWorkerMessageActivitySeq++; + const workerMessage: MockWorkerMessage = { + intentId: id, + message: normalizedMessage, + state: "running", + activitySeq, + }; + mockWorkerMessages.set(normalizedRequestId, workerMessage); + mockActivity.push({ + seq: activitySeq, + intent_id: id, + worker: "user", + ts: new Date().toISOString(), + kind: "user", + summary: normalizedMessage, + detail: normalizedMessage, + } satisfies Activity); + return { + ok: true, + state: workerMessage.state, + activitySeq: workerMessage.activitySeq, + requestId: normalizedRequestId, + }; +} + function controlMockTask(id: string, action: "pause" | "resume"): BatchControlItem { const task = mockTasks.find((item) => item.id === id); if (!task) return { id, ok: false, error: "任务不存在" }; @@ -917,7 +987,7 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re intents: 1, facts: 1, findings: 1, - activities: D.activity.filter((item) => item.intent_id === id).length, + activities: mockActivity.filter((item) => item.intent_id === id).length, }, }; } @@ -926,6 +996,17 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re if (!result.ok) throw new Error(result.error ?? "Worker 状态已变化"); return { id: Number(id.replace(/\D/g, "")) || 0, state: result.state }; } + if (seg[0] === "tasks" && seg[2] === "intents" && seg[4] === "messages" && m === "POST") { + const id = seg[3]; + const result = sendMockWorkerMessage(id, String(b.message ?? ""), String(b.request_id ?? "")); + if (!result.ok) throw new Error(result.error ?? "Worker 状态已变化"); + return { + id: Number(id.replace(/\D/g, "")) || 0, + state: result.state ?? "running", + accepted: true, + request_id: result.requestId ?? "", + }; + } if (seg[0] === "tasks" && seg.length === 3 && seg[2] === "control" && m === "POST") { const action = b.action === "resume" ? "resume" : "pause"; const result = controlMockTask(seg[1], action); @@ -1464,13 +1545,11 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re if (path === "/exploration/activity" && seg.length === 2) { const since = Number(q.get("since") ?? 0); const limit = Math.max(1, Number(q.get("limit") ?? 300)); - const items = D.activityForTask() - .filter((item) => item.seq > since) - .slice(0, limit); + const items = mockActivity.filter((item) => item.seq > since).slice(0, limit); return { items, cursor: items.length ? items[items.length - 1].seq : since }; } if (seg[0] === "exploration" && seg[1] === "activity" && seg.length === 3) { - const a = D.activity.find((x) => x.seq === Number(seg[2])); + const a = mockActivity.find((x) => x.seq === Number(seg[2])); return { detail: a?.detail ?? a?.summary ?? "" }; } if (path === "/tokens/daily") return D.dailyTokens;