Merge pull request #85 from neouks/main

支持 Worker 对话实时调整执行方向

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 = "<!-- ARTEX_WORKER_CHAT:"
+
+func workerChatMarker(requestID string) string {
+	return workerChatMarkerPrefix + requestID + " -->"
+}
+
+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请优先执行这条人工输入。"
 		}
 	}
 

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

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) {

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,
+	})
+}

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)

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

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({
     <Loader2Icon className="size-3.5 animate-spin text-blue-500" />
   ) : 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 (
     <div
       className={cn(
@@ -386,33 +395,8 @@ function SessionItem({
           </span>
         )}
       </button>
-      {controllable && (
+      {cancellable && (
         <div className="flex shrink-0 items-center gap-0.5 pr-1 opacity-100 sm:opacity-0 sm:transition-opacity sm:group-hover/session:opacity-100 sm:group-focus-within/session:opacity-100">
-          {s.status === "running" ? (
-            <Button
-              type="button"
-              variant="ghost"
-              size="icon-xs"
-              onClick={onPause}
-              disabled={controlling}
-              title="暂停 Worker"
-              aria-label="暂停 Worker"
-            >
-              {controlling ? <Loader2Icon className="animate-spin" /> : <PauseIcon />}
-            </Button>
-          ) : (
-            <Button
-              type="button"
-              variant="ghost"
-              size="icon-xs"
-              onClick={onResume}
-              disabled={controlling}
-              title="恢复 Worker"
-              aria-label="恢复 Worker"
-            >
-              {controlling ? <Loader2Icon className="animate-spin" /> : <PlayIcon />}
-            </Button>
-          )}
           <Button
             type="button"
             variant="ghost"
@@ -495,6 +479,9 @@ export function SessionsTab({ taskId }: { taskId: string }) {
   const [controllingIntent, setControllingIntent] = React.useState<string | null>(null);
   const [cancelIntent, setCancelIntent] = React.useState<Session | null>(null);
   const [cancelReason, setCancelReason] = React.useState("");
+  const [workerMessage, setWorkerMessage] = React.useState("");
+  const [workerMessageRequestId, setWorkerMessageRequestId] = React.useState("");
+  const [workerMessageSending, setWorkerMessageSending] = React.useState(false);
   // 方式1 文件上传:选好的附件(已落到任务工作目录 uploads/),随下条消息一起发。
   const [attachments, setAttachments] = React.useState<ChatAttachment[]>([]);
   const [uploading, setUploading] = React.useState(false);
@@ -554,6 +541,7 @@ export function SessionsTab({ taskId }: { taskId: string }) {
     },
     [controllingIntent, patchIntentState, taskId],
   );
+
   // SSE connection state — surfaced so a dropped realtime link is visible, never
   // silently shown as "no messages".
   const [sseLive, setSseLive] = React.useState(false);
@@ -1345,6 +1333,33 @@ export function SessionsTab({ taskId }: { taskId: string }) {
       .finally(() => setSending(false));
   }
 
+  function sendWorkerChat() {
+    const intentId = active.intent_id;
+    const message = workerMessage.trim();
+    if (!intentId || active.inherited || active.status !== "paused" || workerMessageSending || !message) return;
+    if (workerMessageCharCount(message) > MAX_WORKER_MESSAGE_CHARS) {
+      toast.error(`消息不能超过 ${MAX_WORKER_MESSAGE_CHARS} 个字符`);
+      return;
+    }
+    const requestId = workerMessageRequestId || newWorkerMessageRequestID();
+    if (!workerMessageRequestId) setWorkerMessageRequestId(requestId);
+    setWorkerMessageSending(true);
+    api
+      .sendWorkerMessage(taskId, intentId, message, requestId)
+      .then((result) => {
+        // The server records the user turn and streams the run over SSE, so there is
+        // no optimistic insert: the message and the continuation arrive live.
+        patchIntentState(intentId, result.state);
+        setWorkerMessage("");
+        setWorkerMessageRequestId("");
+        toast.success(`消息已发送给 Worker #${intentId}，已立即继续执行`);
+      })
+      .catch((error) => {
+        toast.error(`发送失败：${(error as Error).message || "请稍后重试"}`);
+      })
+      .finally(() => setWorkerMessageSending(false));
+  }
+
   const mainLoaded = !!store.main?.loaded;
   // 发送键位由系统设置决定（localStorage），默认 Enter 发送。
   const sendMode = useChatSendMode();
@@ -1475,11 +1490,11 @@ export function SessionsTab({ taskId }: { taskId: string }) {
                           onClick={() => {
                             setActiveId(s.id);
                             setListOpen(false); // 手机端选完即收起，把高度还给会话记录
+                            setWorkerMessage("");
+                            setWorkerMessageRequestId("");
                           }}
                           controlling={controllingIntent === s.intent_id}
                           deleted={meta?.deleted}
-                          onPause={() => void controlWorker(s, "pause")}
-                          onResume={() => void controlWorker(s, "resume")}
                           onCancel={() => {
                             setCancelIntent(s);
                           }}
@@ -1749,6 +1764,86 @@ export function SessionsTab({ taskId }: { taskId: string }) {
                 </InputGroupAddon>
               </InputGroup>
             </div>
+          ) : active.role === "worker" &&
+            !active.inherited &&
+            (active.status === "running" || active.status === "paused") ? (
+            <div className="border-t p-3">
+              <InputGroup className="min-h-9 has-disabled:opacity-100">
+                <InputGroupTextarea
+                  rows={1}
+                  aria-label={`给 Worker #${active.intent_id} 发消息`}
+                  placeholder={`给 Worker #${active.intent_id} 发消息，调整执行方向…`}
+                  value={workerMessage}
+                  onChange={(event) => {
+                    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"
+                />
+                <InputGroupAddon align="block-end">
+                  <span
+                    className={cn(
+                      "px-1 text-[10px] tabular-nums text-muted-foreground",
+                      workerMessageCharCount(workerMessage) > MAX_WORKER_MESSAGE_CHARS && "text-destructive",
+                    )}
+                  >
+                    {workerMessageCharCount(workerMessage)}/{MAX_WORKER_MESSAGE_CHARS}
+                  </span>
+                  {active.status === "running" ? (
+                    <InputGroupButton
+                      className="ml-auto"
+                      size="icon-xs"
+                      variant="destructive"
+                      onClick={() => void controlWorker(active, "pause")}
+                      disabled={controllingIntent === active.intent_id}
+                      title="暂停当前 Worker"
+                      aria-label="暂停当前 Worker"
+                    >
+                      {controllingIntent === active.intent_id ? (
+                        <Loader2Icon className="animate-spin" />
+                      ) : (
+                        <SquareIcon />
+                      )}
+                    </InputGroupButton>
+                  ) : (
+                    <>
+                      <InputGroupButton
+                        className="ml-auto"
+                        size="xs"
+                        variant="ghost"
+                        onClick={() => void controlWorker(active, "resume")}
+                        disabled={controllingIntent === active.intent_id || workerMessageSending}
+                        title="不发消息，直接继续执行"
+                        aria-label="直接继续执行"
+                      >
+                        {controllingIntent === active.intent_id ? <Loader2Icon className="animate-spin" /> : "直接继续"}
+                      </InputGroupButton>
+                      <InputGroupButton
+                        size="icon-xs"
+                        variant="default"
+                        onClick={sendWorkerChat}
+                        disabled={
+                          workerMessageSending ||
+                          !workerMessage.trim() ||
+                          workerMessageCharCount(workerMessage) > MAX_WORKER_MESSAGE_CHARS
+                        }
+                        title="发送消息"
+                        aria-label="发送消息"
+                      >
+                        {workerMessageSending ? <Spinner /> : <ArrowUpIcon />}
+                      </InputGroupButton>
+                    </>
+                  )}
+                </InputGroupAddon>
+              </InputGroup>
+            </div>
           ) : (
             <div className="flex items-center border-t px-4 py-2">
               <TodoPopover

web/src/lib/api.ts

@@ -284,6 +284,13 @@ export const api = {
       state: "paused" | "open" | "stopped";
       deleted?: { intents: number; facts: number; findings: number; activities: number };
     }>(`/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<TaskLLMResolutions>(`/tasks/${id}/llm/resolution`),
   // 重跑一条没跑成功的意图(blocked/exhausted/stopped)：置回 open，worker 会重新认领、从头再跑。
   rerunIntent: (taskId: string, intentId: string) =>

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<string, MockWorkerMessage>();
+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;