From c9bf534818e38033a2f0d276454ccfa8c8a2b777 Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Mon, 31 Aug 2026 09:00:21 -0400 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=20worker=20=E5=AF=B9?= =?UTF-8?q?=E8=AF=9D=EF=BC=9A=E5=A4=8D=E7=94=A8=20resume=20=E6=9C=BA?= =?UTF-8?q?=E5=88=B6=EF=BC=8C=E6=94=B9=E4=B8=93=E5=B1=9E=20goroutine=20?= =?UTF-8?q?=E8=BF=90=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把「给运行中 Worker 发消息调方向」从自建的介入持久化协议，改为复用 已有的暂停/恢复 + transcript resume 能力，实现与主 Agent chat 一致。 - 拆 runWorkerStep：抽出 runIntent(执行+善后)，池子与消息路径共用一份 善后逻辑；新增 runDetachedIntent，把意图 paused→running(不经 frontier) 并在专属 goroutine 里直接跑，用户消息不受 3 槽池限制、随发随跑。 - 精简 sendWorkerMessage handler：校验 + emit 用户活动 + 专属运行；消息经 ExecuteWithMessage 作为下一轮输入注入，request_id 仅做 transcript 去重。 - 删除介入 DB 协议与 15 处 activity 查询过滤(回退 exploration.go)、 directedClaims 调度屏障、RecoverWorkerMessages 恢复、workerWake/waitWorker、 AbortWorkIntervenedByUser。无 DB schema 改动。 - 前端：保留消息框，恢复「直接继续」(resume)按钮；发送改由 SSE 承载， 去掉乐观插入。 - 取舍(低频可接受)：消息仅存内存，不做崩溃恢复；发消息时该任务可能瞬时 超并发 1 个 worker。 净改动 9 文件 +583/-46(原 PR 15 文件 +1940/-72)。 --- agent/cancelcause.go | 2 - db/exploration.go | 412 +------------ db/intent_intervention_lifecycle_test.go | 204 ------- server/engine.go | 559 +++--------------- server/engine_cancelcause_test.go | 1 - server/goals.go | 7 - server/intent_intervention.go | 65 +- server/intent_intervention_protocol_test.go | 378 ------------ server/manager.go | 17 +- server/server.go | 8 - server/worker_message_test.go | 54 ++ .../tasks/detail/_tabs/sessions-tab.tsx | 72 +-- web/src/lib/api.ts | 3 +- web/src/lib/mock/handler.ts | 23 +- 14 files changed, 237 insertions(+), 1568 deletions(-) delete mode 100644 db/intent_intervention_lifecycle_test.go delete mode 100644 server/intent_intervention_protocol_test.go create mode 100644 server/worker_message_test.go diff --git a/agent/cancelcause.go b/agent/cancelcause.go index e61cfaf..f4705bc 100644 --- a/agent/cancelcause.go +++ b/agent/cancelcause.go @@ -45,8 +45,6 @@ var ( "规划者调用 kill_work 主动终止了这条意图，通常表示方向跑偏或已无继续价值；意图会标记为 stopped，不会自动重新领取") AbortWorkPausedByUser = cause("work_paused_by_user", "用户暂停了这条 Worker 意图", "用户暂停了正在运行的 Worker。本次调用被取消，意图转为 paused；已经登记的意图、事实、漏洞和活动记录全部保留，恢复后从头重新执行") - AbortWorkIntervenedByUser = cause("work_intervened_by_user", "用户暂停 Worker 并发起了新对话", - "用户向正在运行的 Worker 发送了新的对话意图。本次调用被立即取消并短暂进入 paused；新消息作为同一会话的下一条用户消息，随后自动恢复并优先重新领取") AbortWorkCancelledByUser = cause("work_cancelled_by_user", "用户取消了这条 Worker 意图", "用户取消了正在运行的 Worker。本次调用被取消；Worker 退出写入区后，服务端会事务性删除该意图及其直接产生的事实、漏洞和执行记录") AbortWorkFinished = cause("work_finished", "Worker 已正常结束并释放 context", diff --git a/db/exploration.go b/db/exploration.go index 1ffb92a..eb0e677 100644 --- a/db/exploration.go +++ b/db/exploration.go @@ -14,11 +14,6 @@ import ( // it to distinguish an expected control race (HTTP 409) from a storage failure. var ErrIntentStateConflict = errors.New("intent state changed concurrently") -// ErrIntentInterventionConflict marks a rejected or conflicting worker -// intervention request. It covers state races, a reused request id with a -// different payload, and a second request trying to overtake a pending one. -var ErrIntentInterventionConflict = errors.New("intent intervention conflict") - // utf8Clean makes a string safe for a PostgreSQL text column. It (1) replaces // invalid UTF-8 byte sequences with U+FFFD and (2) strips NUL (0x00) bytes. // Tool output (nmap/curl/… stdout) can carry raw/truncated bytes that PostgreSQL's @@ -335,376 +330,6 @@ WHERE id=$2 AND exploration_id=$3 AND kind='intent' AND state=$4`, state, id, s. return n == 1, err } -const ( - IntentInterventionPending = "pending" - IntentInterventionAccepted = "accepted" - IntentInterventionRejected = "rejected" -) - -// IntentIntervention is the durable control record for one human interruption. -// It deliberately lives in activity rather than a new table: activity already -// participates in task archive/restore, and the accepted row is also the user -// turn rendered in the intent transcript. -type IntentIntervention struct { - RequestID string - Message string - Status string - ActivityID int64 - IntentID int64 - IntentState string - CreatedAt time.Time -} - -// lockIntentIntervention serializes request-id reservation/finalization across -// processes. exploration_nodes.id is a global primary key, so the intent id is -// sufficient input to the namespaced advisory key. -func lockIntentIntervention(tx *sql.Tx, intentID int64) error { - _, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtextextended('artex-intent-intervention:' || $1::text, 0))`, intentID) - return err -} - -func scanIntentIntervention(row *sql.Row) (*IntentIntervention, error) { - var it IntentIntervention - err := row.Scan(&it.ActivityID, &it.RequestID, &it.Message, &it.Status, - &it.IntentID, &it.IntentState, &it.CreatedAt) - if err == sql.ErrNoRows { - return nil, nil - } - if err != nil { - return nil, err - } - return &it, nil -} - -// IntentInterventionByRequest finds a prior reservation/acceptance. Callers do -// this before ordinary lifecycle validation so retrying an already accepted -// request remains idempotent even if the worker has since finished naturally. -func (s *ExplorationStore) IntentInterventionByRequest(intentID int64, requestID string) (*IntentIntervention, error) { - return scanIntentIntervention(s.db.QueryRow(` -SELECT COALESCE(NULLIF(a.metadata->>'accepted_activity_id','')::bigint, a.id), - COALESCE(a.metadata->>'intervention_request_id',''), - COALESCE(a.metadata->>'intervention_message', a.detail, ''), - COALESCE(a.metadata->>'intervention_status',''), - n.id, n.state, a.created_at -FROM activity a -JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id -WHERE a.exploration_id=$1 AND a.node_id=$2 - AND a.metadata->>'intervention_request_id'=$3 -ORDER BY a.id -LIMIT 1`, s.expID, intentID, requestID)) -} - -// ReserveIntentIntervention durably claims a request id while the intent is -// still running. A hidden intervention_control activity is the write-ahead -// record; acceptance inserts a newer visible kind='user' row so SSE cursors can -// never skip a message whose control reservation had an older id. -func (s *ExplorationStore) ReserveIntentIntervention(intentID int64, requestID, message string) (*IntentIntervention, error) { - tx, err := s.db.Begin() - if err != nil { - return nil, err - } - defer tx.Rollback() - if err := lockIntentIntervention(tx, intentID); err != nil { - return nil, err - } - - existing, err := scanIntentIntervention(tx.QueryRow(` -SELECT COALESCE(NULLIF(a.metadata->>'accepted_activity_id','')::bigint, a.id), - COALESCE(a.metadata->>'intervention_request_id',''), - COALESCE(a.metadata->>'intervention_message', a.detail, ''), - COALESCE(a.metadata->>'intervention_status',''), - n.id, n.state, a.created_at -FROM activity a -JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id -WHERE a.exploration_id=$1 AND a.node_id=$2 - AND a.metadata->>'intervention_request_id'=$3 -ORDER BY a.id -LIMIT 1`, s.expID, intentID, requestID)) - if err != nil { - return nil, err - } - if existing != nil { - if existing.Message != message { - return nil, fmt.Errorf("%w: request_id 已用于不同的 Worker 消息", ErrIntentInterventionConflict) - } - if err := tx.Commit(); err != nil { - return nil, err - } - return existing, nil - } - - var state string - if err := tx.QueryRow(`SELECT state FROM exploration_nodes - WHERE id=$1 AND exploration_id=$2 AND kind='intent' FOR UPDATE`, intentID, s.expID).Scan(&state); err != nil { - if err == sql.ErrNoRows { - return nil, fmt.Errorf("%w: intent not found", ErrIntentInterventionConflict) - } - return nil, err - } - if state != "running" && state != "paused" { - return nil, fmt.Errorf("%w: intent state %s cannot receive a message", ErrIntentInterventionConflict, state) - } - var otherRequest string - err = tx.QueryRow(`SELECT COALESCE(metadata->>'intervention_request_id','') - FROM activity - WHERE exploration_id=$1 AND node_id=$2 - AND metadata->>'intervention_status'=$3 - ORDER BY id LIMIT 1`, s.expID, intentID, IntentInterventionPending).Scan(&otherRequest) - if err != nil && err != sql.ErrNoRows { - return nil, err - } - if err == nil { - return nil, fmt.Errorf("%w: intent 已有待处理的 Worker 消息 %s", ErrIntentInterventionConflict, otherRequest) - } - - metadata, err := json.Marshal(map[string]any{ - "intervention": true, - "intervention_handoff": "pending", - "intervention_request_id": requestID, - "intervention_message": message, - "intervention_status": IntentInterventionPending, - "source": "user", - }) - if err != nil { - return nil, err - } - var out IntentIntervention - err = tx.QueryRow(` -INSERT INTO activity(exploration_id,node_id,worker,kind,metadata) -VALUES ($1,$2,'system','intervention_control',$3) -RETURNING id, created_at`, s.expID, intentID, string(metadata)).Scan(&out.ActivityID, &out.CreatedAt) - if err != nil { - return nil, err - } - out.RequestID, out.Message, out.Status = requestID, message, IntentInterventionPending - out.IntentID, out.IntentState = intentID, state - if err := tx.Commit(); err != nil { - return nil, err - } - return &out, nil -} - -// AcceptIntentIntervention is the DB commit point: the hidden reservation -// becomes an audited user turn in the same transaction that moves paused back -// to open. Thus no Worker can claim the intent before its pending message and -// visible activity are both durable. -func (s *ExplorationStore) AcceptIntentIntervention(intentID int64, requestID, message string) (Activity, bool, error) { - var activity Activity - tx, err := s.db.Begin() - if err != nil { - return activity, false, err - } - defer tx.Rollback() - if err := lockIntentIntervention(tx, intentID); err != nil { - return activity, false, err - } - - var controlID int64 - var storedMessage, status, state, taskStatus string - var taskPaused, taskQueued bool - var taskDeletedAt sql.NullTime - var createdAt time.Time - err = tx.QueryRow(` -SELECT a.id, - COALESCE(a.metadata->>'intervention_message', a.detail, ''), - COALESCE(a.metadata->>'intervention_status',''), - n.state, a.created_at, COALESCE(t.status,''), t.paused, t.queued, t.deleted_at -FROM activity a -JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id -JOIN tasks t ON t.exploration_id=a.exploration_id -WHERE a.exploration_id=$1 AND a.node_id=$2 - AND a.metadata->>'intervention_request_id'=$3 - ORDER BY a.id LIMIT 1 - FOR UPDATE OF a,n,t`, s.expID, intentID, requestID). - Scan(&controlID, &storedMessage, &status, &state, &createdAt, - &taskStatus, &taskPaused, &taskQueued, &taskDeletedAt) - if err == sql.ErrNoRows { - return activity, false, fmt.Errorf("%w: intervention reservation not found", ErrIntentInterventionConflict) - } - if err != nil { - return activity, false, err - } - if storedMessage != message { - return activity, false, fmt.Errorf("%w: request_id 已用于不同的 Worker 消息", ErrIntentInterventionConflict) - } - if status == IntentInterventionAccepted { - var acceptedID int64 - if err := tx.QueryRow(`SELECT COALESCE((metadata->>'accepted_activity_id')::bigint,0), - COALESCE((SELECT created_at FROM activity accepted - WHERE accepted.id=(activity.metadata->>'accepted_activity_id')::bigint), created_at) - FROM activity WHERE id=$1`, controlID).Scan(&acceptedID, &createdAt); err != nil { - return activity, false, err - } - if acceptedID <= 0 { - return activity, false, fmt.Errorf("accepted intervention has no activity id") - } - if err := tx.Commit(); err != nil { - return activity, false, err - } - nid := intentID - activity.ID = acceptedID - activity.NodeID, activity.Worker, activity.Kind = &nid, "user", "user" - activity.Summary, activity.Detail, activity.CreatedAt = message, message, createdAt - return activity, true, nil - } - if status != IntentInterventionPending { - return activity, false, fmt.Errorf("%w: intervention status %s cannot be accepted", ErrIntentInterventionConflict, status) - } - if state != "paused" { - return activity, false, fmt.Errorf("%w: intent state %s cannot be reopened", ErrIntentInterventionConflict, state) - } - if taskPaused || taskQueued || taskDeletedAt.Valid || IsTerminal(taskStatus) { - return activity, false, fmt.Errorf("%w: task lifecycle changed (status=%s paused=%v queued=%v deleted=%v)", - ErrIntentInterventionConflict, taskStatus, taskPaused, taskQueued, taskDeletedAt.Valid) - } - - metadata, err := json.Marshal(map[string]any{ - "intervention": true, - "intervention_request_id": requestID, - "intervention_message": message, - "intervention_status": IntentInterventionAccepted, - "source": "user", - }) - if err != nil { - return activity, false, err - } - if err := tx.QueryRow(`INSERT INTO activity( - exploration_id,node_id,worker,kind,summary,detail,metadata) - VALUES ($1,$2,'user','user',$3,$3,$4) - RETURNING id,created_at`, s.expID, intentID, utf8Clean(message), string(metadata)). - Scan(&activity.ID, &createdAt); err != nil { - return activity, false, err - } - var acceptedMeta map[string]any - if err := json.Unmarshal(metadata, &acceptedMeta); err != nil { - return activity, false, err - } - acceptedMeta["accepted_activity_id"] = activity.ID - controlMetadata, err := json.Marshal(acceptedMeta) - if err != nil { - return activity, false, err - } - if _, err := tx.Exec(`UPDATE activity SET metadata=$1 WHERE id=$2`, string(controlMetadata), controlID); err != nil { - return activity, false, err - } - res, err := tx.Exec(`UPDATE exploration_nodes - SET state='open', owner=NULL, completed_at=NULL, blocked_reason=NULL - WHERE id=$1 AND exploration_id=$2 AND kind='intent' AND state='paused'`, intentID, s.expID) - if err != nil { - return activity, false, err - } - if n, err := res.RowsAffected(); err != nil || n != 1 { - if err != nil { - return activity, false, err - } - return activity, false, fmt.Errorf("%w: intent is no longer paused", ErrIntentInterventionConflict) - } - if err := tx.Commit(); err != nil { - return activity, false, err - } - nid := intentID - activity.NodeID, activity.Worker, activity.Kind = &nid, "user", "user" - activity.Summary, activity.Detail, activity.CreatedAt = message, message, createdAt - activity.Metadata = metadata - return activity, false, nil -} - -// RecoverableIntentInterventions returns control operations that must be -// reconstructed before ordinary frontier claiming after a process restart: -// pending transcript commits, plus accepted hand-offs not yet claimed once. -func (s *ExplorationStore) RecoverableIntentInterventions() ([]IntentIntervention, error) { - rows, err := s.db.Query(` -SELECT COALESCE(NULLIF(a.metadata->>'accepted_activity_id','')::bigint, a.id), - COALESCE(a.metadata->>'intervention_request_id',''), - COALESCE(a.metadata->>'intervention_message',''), - COALESCE(a.metadata->>'intervention_status',''), - n.id,n.state,a.created_at -FROM activity a -JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id -WHERE a.exploration_id=$1 AND a.kind='intervention_control' - AND (a.metadata->>'intervention_status'=$2 OR ( - a.metadata->>'intervention_status'=$3 - AND COALESCE(a.metadata->>'intervention_handoff','pending')='pending')) -ORDER BY a.id`, s.expID, IntentInterventionPending, IntentInterventionAccepted) - if err != nil { - return nil, err - } - defer rows.Close() - out := []IntentIntervention{} - for rows.Next() { - var item IntentIntervention - if err := rows.Scan(&item.ActivityID, &item.RequestID, &item.Message, &item.Status, - &item.IntentID, &item.IntentState, &item.CreatedAt); err != nil { - return nil, err - } - out = append(out, item) - } - return out, rows.Err() -} - -// MarkIntentInterventionHandoffClaimed records that the accepted instruction -// has been preferentially claimed once. Idempotency status remains accepted. -func (s *ExplorationStore) MarkIntentInterventionHandoffClaimed(intentID int64) error { - _, err := s.db.Exec(`UPDATE activity SET metadata=jsonb_set( - metadata,'{intervention_handoff}',to_jsonb('claimed'::text)) - WHERE id=(SELECT id FROM activity - WHERE exploration_id=$1 AND node_id=$2 AND kind='intervention_control' - AND metadata->>'intervention_status'=$3 - AND COALESCE(metadata->>'intervention_handoff','pending')='pending' - ORDER BY id DESC LIMIT 1)`, s.expID, intentID, IntentInterventionAccepted) - return err -} - -// PendingIntentInterventionMessage returns the accepted human chat turn that -// should be consumed by the next Worker run. The handoff remains pending until -// the run has actually started, so a process restart can reconstruct the -// directed claim without losing the user's message. -func (s *ExplorationStore) PendingIntentInterventionMessage(intentID int64) (string, string, bool, error) { - var requestID, message string - err := s.db.QueryRow(`SELECT - COALESCE(metadata->>'intervention_request_id',''), - COALESCE(metadata->>'intervention_message','') - FROM activity - WHERE exploration_id=$1 AND node_id=$2 AND kind='intervention_control' - AND metadata->>'intervention_status'=$3 - AND COALESCE(metadata->>'intervention_handoff','pending')='pending' - ORDER BY id DESC LIMIT 1`, s.expID, intentID, IntentInterventionAccepted). - Scan(&requestID, &message) - if err == sql.ErrNoRows { - return "", "", false, nil - } - if err != nil { - return "", "", false, err - } - return requestID, message, true, nil -} - -// RejectIntentIntervention closes a pending control reservation without making -// it visible in task activity. It is used when lifecycle validation changes or -// the transcript could not be made durable; a different future request is then -// free to proceed after the user explicitly resumes the paused intent. -func (s *ExplorationStore) RejectIntentIntervention(intentID int64, requestID, reason string) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - if err := lockIntentIntervention(tx, intentID); err != nil { - return err - } - _, err = tx.Exec(`UPDATE activity - SET metadata=jsonb_set(jsonb_set(metadata,'{intervention_status}',to_jsonb($1::text)), - '{intervention_reject_reason}',to_jsonb($2::text)) - WHERE exploration_id=$3 AND node_id=$4 - AND metadata->>'intervention_request_id'=$5 - AND metadata->>'intervention_status'=$6`, - IntentInterventionRejected, utf8Clean(reason), s.expID, intentID, requestID, IntentInterventionPending) - if err != nil { - return err - } - return tx.Commit() -} - // IntentCleanup summarizes the blackboard records removed when a user cancels // one worker intent. Global assets and traffic are deliberately not part of this // operation; exploration anchors disappear only because their owning nodes do. @@ -1381,8 +1006,7 @@ func (d *DB) TokenTotalsAll() (map[int64]TokenUsage, error) { // Persisted (unlike Engine.LastActivity's in-memory map), so it survives restarts // and gives终态任务 a stable "ran until" time for computing run duration. func (d *DB) LastActivityAll() (map[int64]int64, error) { - rows, err := d.Query(`SELECT exploration_id, EXTRACT(EPOCH FROM MAX(created_at))::bigint - FROM activity WHERE kind IS DISTINCT FROM 'intervention_control' GROUP BY exploration_id`) + rows, err := d.Query(`SELECT exploration_id, EXTRACT(EPOCH FROM MAX(created_at))::bigint FROM activity GROUP BY exploration_id`) if err != nil { return nil, err } @@ -1433,7 +1057,7 @@ func (d *DB) TaskListMetricsAll() (map[int64]TaskListMetrics, error) { LEFT JOIN LATERAL ( SELECT EXTRACT(EPOCH FROM created_at)::bigint AS created_at FROM activity - WHERE exploration_id=task.exploration_id AND kind IS DISTINCT FROM 'intervention_control' + WHERE exploration_id=task.exploration_id ORDER BY created_at DESC LIMIT 1 ) latest_activity ON true @@ -1564,10 +1188,10 @@ func (s *ExplorationStore) ActivityList(nodeID *int64, sinceID int64, limit int) const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens` if nodeID != nil { rows, err = s.db.Query(`SELECT `+cols+` -FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind IS DISTINCT FROM 'intervention_control' AND id>$3 ORDER BY id LIMIT $4`, s.expID, *nodeID, sinceID, limit) +FROM activity WHERE exploration_id=$1 AND node_id=$2 AND id>$3 ORDER BY id LIMIT $4`, s.expID, *nodeID, sinceID, limit) } else { rows, err = s.db.Query(`SELECT `+cols+` -FROM activity WHERE exploration_id=$1 AND kind IS DISTINCT FROM 'intervention_control' AND id>$2 ORDER BY id LIMIT $3`, s.expID, sinceID, limit) +FROM activity WHERE exploration_id=$1 AND id>$2 ORDER BY id LIMIT $3`, s.expID, sinceID, limit) } if err != nil { return nil, sinceID, err @@ -1604,7 +1228,7 @@ func (s *ExplorationStore) ActivityListForTerminalIntent(nodeID, sinceID int64, JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id WHERE a.exploration_id=$1 AND a.node_id=$2 AND a.id>$3 AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped') - AND a.kind NOT IN ('thinking','usage','intervention_control') + AND a.kind NOT IN ('thinking','usage') ORDER BY a.id LIMIT $4`, s.expID, nodeID, sinceID, limit) if err != nil { return nil, sinceID, err @@ -1667,7 +1291,7 @@ func (s *ExplorationStore) ActivityPage(f ActivitySessionFilter, before int64, l limitArg := len(args) + 1 args = append(args, limit+1) // one extra row to detect older history q := fmt.Sprintf(`SELECT `+cols+` -FROM activity WHERE exploration_id=$1 AND kind IS DISTINCT FROM 'intervention_control'%s AND ($%d <= 0 OR id < $%d) +FROM activity WHERE exploration_id=$1%s AND ($%d <= 0 OR id < $%d) ORDER BY id DESC LIMIT $%d`, cond, beforeArg, beforeArg, limitArg) rows, err := s.db.Query(q, args...) if err != nil { @@ -1712,7 +1336,7 @@ func (s *ExplorationStore) ActivityPageForTerminalIntent(nodeID, before int64, l JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id WHERE a.exploration_id=$1 AND a.node_id=$2 AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped') - AND a.kind NOT IN ('thinking','usage','intervention_control') + AND a.kind NOT IN ('thinking','usage') AND ($3 <= 0 OR a.id < $3) ORDER BY a.id DESC LIMIT $4`, s.expID, nodeID, before, limit+1) if err != nil { @@ -1748,7 +1372,7 @@ func (s *ExplorationStore) ActivityPageForTerminalIntent(nodeID, before int64, l // session, since one SSE covers the whole task. func (s *ExplorationStore) ActivityMaxID() (int64, error) { var max sql.NullInt64 - err := s.db.QueryRow(`SELECT MAX(id) FROM activity WHERE exploration_id=$1 AND kind IS DISTINCT FROM 'intervention_control'`, s.expID).Scan(&max) + err := s.db.QueryRow(`SELECT MAX(id) FROM activity WHERE exploration_id=$1`, s.expID).Scan(&max) if err != nil { return 0, err } @@ -1758,7 +1382,7 @@ func (s *ExplorationStore) ActivityMaxID() (int64, error) { // ActivityDetail lazily returns the full detail blob for one step. func (s *ExplorationStore) ActivityDetail(id int64) (string, error) { var d sql.NullString - err := s.db.QueryRow(`SELECT detail FROM activity WHERE id=$1 AND exploration_id=$2 AND kind IS DISTINCT FROM 'intervention_control'`, id, s.expID).Scan(&d) + err := s.db.QueryRow(`SELECT detail FROM activity WHERE id=$1 AND exploration_id=$2`, id, s.expID).Scan(&d) if err == sql.ErrNoRows { return "", nil } @@ -1791,7 +1415,7 @@ func (s *ExplorationStore) ActivityTrace(nodeID int64, limit int) ([]Activity, e limit = 500 } rows, err := s.db.Query(`SELECT `+traceCols+` -FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage','intervention_control') +FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage') ORDER BY id LIMIT $3`, s.expID, nodeID, limit) if err != nil { return nil, err @@ -1811,7 +1435,7 @@ func (s *ExplorationStore) ActivityTraceForTerminalIntent(nodeID int64, limit in JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id WHERE a.exploration_id=$1 AND a.node_id=$2 AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped') - AND a.kind NOT IN ('thinking','usage','intervention_control') + AND a.kind NOT IN ('thinking','usage') ORDER BY a.id LIMIT $3`, s.expID, nodeID, limit) if err != nil { return nil, err @@ -1833,11 +1457,11 @@ func (s *ExplorationStore) ActivityTraceSearch(nodeID *int64, q string, limit in var err error if nodeID != nil { rows, err = s.db.Query(`SELECT `+traceCols+` -FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage','intervention_control') +FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage') AND (summary ILIKE $3 OR detail ILIKE $3) ORDER BY id LIMIT $4`, s.expID, *nodeID, like, limit) } else { rows, err = s.db.Query(`SELECT `+traceCols+` -FROM activity WHERE exploration_id=$1 AND node_id IS NOT NULL AND kind NOT IN ('thinking','usage','intervention_control') +FROM activity WHERE exploration_id=$1 AND node_id IS NOT NULL AND kind NOT IN ('thinking','usage') AND (summary ILIKE $2 OR detail ILIKE $2) ORDER BY id LIMIT $3`, s.expID, like, limit) } if err != nil { @@ -1859,7 +1483,7 @@ func (s *ExplorationStore) ActivityTraceSearchForTerminalIntent(nodeID int64, q JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id WHERE a.exploration_id=$1 AND a.node_id=$2 AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped') - AND a.kind NOT IN ('thinking','usage','intervention_control') + AND a.kind NOT IN ('thinking','usage') AND (a.summary ILIKE $3 OR a.detail ILIKE $3) ORDER BY a.id LIMIT $4`, s.expID, nodeID, like, limit) if err != nil { @@ -1881,7 +1505,7 @@ func (s *ExplorationStore) ActivityTraceSearchTerminalIntents(q string, limit in JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id WHERE a.exploration_id=$1 AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped') - AND a.kind NOT IN ('thinking','usage','intervention_control') + AND a.kind NOT IN ('thinking','usage') AND (a.summary ILIKE $2 OR a.detail ILIKE $2) ORDER BY a.id LIMIT $3`, s.expID, like, limit) if err != nil { @@ -1952,7 +1576,7 @@ func (s *ExplorationStore) ActivityTraceSearchExcluding(excludeNodeID int64, q s like := "%" + q + "%" rows, err := s.db.Query(`SELECT `+traceCols+` FROM activity WHERE exploration_id=$1 AND node_id IS NOT NULL AND node_id <> $2 -AND kind NOT IN ('thinking','usage','intervention_control') +AND kind NOT IN ('thinking','usage') AND (summary ILIKE $3 OR detail ILIKE $3) ORDER BY id LIMIT $4`, s.expID, excludeNodeID, like, limit) if err != nil { return nil, err @@ -1976,7 +1600,7 @@ func (s *ExplorationStore) ActivityByIDs(ids []int64) ([]Activity, error) { args[i+1] = id } rows, err := s.db.Query(`SELECT id, node_id, COALESCE(kind,''), COALESCE(tool,''), is_error, COALESCE(detail,'') -FROM activity WHERE exploration_id=$1 AND kind NOT IN ('thinking','intervention_control') AND id IN (`+strings.Join(ph, ",")+`) ORDER BY id`, args...) +FROM activity WHERE exploration_id=$1 AND kind<>'thinking' AND id IN (`+strings.Join(ph, ",")+`) ORDER BY id`, args...) if err != nil { return nil, err } @@ -2010,7 +1634,7 @@ func (s *ExplorationStore) ActivityByIDsForTerminalIntents(ids []int64) ([]Activ rows, err := s.db.Query(`SELECT a.id, a.node_id, COALESCE(a.kind,''), COALESCE(a.tool,''), a.is_error, COALESCE(a.detail,'') FROM activity a JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id - WHERE a.exploration_id=$1 AND a.kind NOT IN ('thinking','usage','intervention_control') AND n.kind='intent' + WHERE a.exploration_id=$1 AND a.kind NOT IN ('thinking','usage') AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped') AND a.id IN (`+strings.Join(ph, ",")+`) ORDER BY a.id`, args...) if err != nil { diff --git a/db/intent_intervention_lifecycle_test.go b/db/intent_intervention_lifecycle_test.go deleted file mode 100644 index c27a746..0000000 --- a/db/intent_intervention_lifecycle_test.go +++ /dev/null @@ -1,204 +0,0 @@ -package db - -import ( - "encoding/json" - "errors" - "sync" - "testing" -) - -func newRunningInterventionIntent(t *testing.T, d *DB, label string) (*ExplorationStore, int64, func()) { - t.Helper() - task, err := d.CreateTask("worker intervention "+label, "verify intervention protocol", nil, 0, 0) - if err != nil { - t.Fatal(err) - } - store := d.Exploration(task.ExplorationID) - intentID, err := store.AddIntent(map[string]any{"summary": label}, 1, nil, "planner") - if err != nil { - _ = d.DeleteTask(task.ID) - t.Fatal(err) - } - if claimed, err := store.ClaimIntent(intentID, "work#1"); err != nil || !claimed { - _ = d.DeleteTask(task.ID) - t.Fatalf("claim intervention intent: claimed=%v err=%v", claimed, err) - } - return store, intentID, func() { _ = d.DeleteTask(task.ID) } -} - -func TestIntentInterventionReservationAndAcceptanceAreIdempotent(t *testing.T) { - d, err := Open(testDSN(t)) - if err != nil { - t.Skipf("postgres unavailable (%v) - skipping", err) - } - defer d.Close() - store, intentID, cleanup := newRunningInterventionIntent(t, d, "idempotent") - defer cleanup() - - const requestID = "intervention-idempotent-1" - const message = "停止枚举，优先验证登录越权" - reserved, err := store.ReserveIntentIntervention(intentID, requestID, message) - if err != nil { - t.Fatal(err) - } - if reserved.Status != IntentInterventionPending || reserved.ActivityID <= 0 || reserved.IntentState != "running" { - t.Fatalf("unexpected reservation: %+v", reserved) - } - - replayed, err := store.ReserveIntentIntervention(intentID, requestID, message) - if err != nil { - t.Fatal(err) - } - if replayed.ActivityID != reserved.ActivityID || replayed.Status != IntentInterventionPending { - t.Fatalf("reservation retry created a different operation: first=%+v retry=%+v", reserved, replayed) - } - if _, err := store.ReserveIntentIntervention(intentID, requestID, "改成另一条 Worker 消息"); !errors.Is(err, ErrIntentInterventionConflict) { - t.Fatalf("same request_id with different message error=%v, want intervention conflict", err) - } - if _, err := store.ReserveIntentIntervention(intentID, "intervention-other", message); !errors.Is(err, ErrIntentInterventionConflict) { - t.Fatalf("second pending request error=%v, want intervention conflict", err) - } - - if changed, err := store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed { - t.Fatalf("settle intent before acceptance: changed=%v err=%v", changed, err) - } - activity, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message) - if err != nil || duplicate { - t.Fatalf("accept intervention: duplicate=%v err=%v", duplicate, err) - } - if activity.ID <= reserved.ActivityID || activity.NodeID == nil || *activity.NodeID != intentID || - activity.Worker != "user" || activity.Kind != "user" || activity.Summary != message || activity.Detail != message { - t.Fatalf("accepted audit mismatch: %+v", activity) - } - var metadata map[string]any - if err := json.Unmarshal(activity.Metadata, &metadata); err != nil { - t.Fatal(err) - } - if metadata["intervention_request_id"] != requestID || metadata["intervention_status"] != IntentInterventionAccepted { - t.Fatalf("accepted metadata=%v", metadata) - } - node, err := store.GetNode(intentID) - if err != nil || node == nil || node.State != "open" || node.Owner != "" { - t.Fatalf("accepted intent=%+v err=%v, want reopened and unowned", node, err) - } - - again, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message) - if err != nil || !duplicate || again.ID != activity.ID { - t.Fatalf("accept retry: activity=%+v duplicate=%v err=%v", again, duplicate, err) - } - lookup, err := store.IntentInterventionByRequest(intentID, requestID) - if err != nil || lookup == nil || lookup.Status != IntentInterventionAccepted || lookup.ActivityID != activity.ID { - t.Fatalf("accepted lookup=%+v err=%v", lookup, err) - } - var rows int - if err := d.QueryRow(`SELECT COUNT(*) FROM activity - WHERE exploration_id=$1 AND node_id=$2 - AND metadata->>'intervention_request_id'=$3`, store.ID(), intentID, requestID).Scan(&rows); err != nil { - t.Fatal(err) - } - if rows != 2 { - t.Fatalf("Worker message rows=%d, want one hidden control row and one visible user row", rows) - } - activities, cursor, err := store.ActivityList(&intentID, 0, 10) - if err != nil { - t.Fatal(err) - } - if len(activities) != 1 || activities[0].ID != activity.ID || cursor != activity.ID { - t.Fatalf("public activities leaked control row: activities=%+v cursor=%d", activities, cursor) - } - maxID, err := store.ActivityMaxID() - if err != nil || maxID != activity.ID { - t.Fatalf("public activity cursor=%d err=%v, want visible user row %d", maxID, err, activity.ID) - } - if detail, err := store.ActivityDetail(reserved.ActivityID); err != nil || detail != "" { - t.Fatalf("hidden control detail=%q err=%v", detail, err) - } -} - -func TestIntentInterventionConcurrentReservationHasSingleDurableWinner(t *testing.T) { - d, err := Open(testDSN(t)) - if err != nil { - t.Skipf("postgres unavailable (%v) - skipping", err) - } - defer d.Close() - store, intentID, cleanup := newRunningInterventionIntent(t, d, "concurrent") - defer cleanup() - - const contenders = 12 - const requestID = "intervention-concurrent-1" - const message = "只保留被动请求并验证当前假设" - start := make(chan struct{}) - results := make(chan *IntentIntervention, contenders) - errs := make(chan error, contenders) - var wg sync.WaitGroup - for range contenders { - wg.Add(1) - go func() { - defer wg.Done() - <-start - reservation, reserveErr := store.ReserveIntentIntervention(intentID, requestID, message) - results <- reservation - errs <- reserveErr - }() - } - close(start) - wg.Wait() - close(results) - close(errs) - for err := range errs { - if err != nil { - t.Fatalf("concurrent reservation: %v", err) - } - } - var activityID int64 - for result := range results { - if result == nil || result.ActivityID <= 0 { - t.Fatalf("invalid concurrent reservation: %+v", result) - } - if activityID == 0 { - activityID = result.ActivityID - } else if result.ActivityID != activityID { - t.Fatalf("concurrent requests produced activity ids %d and %d", activityID, result.ActivityID) - } - } - var rows int - if err := d.QueryRow(`SELECT COUNT(*) FROM activity - WHERE exploration_id=$1 AND node_id=$2 - AND metadata->>'intervention_request_id'=$3`, store.ID(), intentID, requestID).Scan(&rows); err != nil { - t.Fatal(err) - } - if rows != 1 { - t.Fatalf("concurrent intervention rows=%d, want 1", rows) - } -} - -func TestIntentInterventionAcceptanceCanRetryAfterStateSettlement(t *testing.T) { - d, err := Open(testDSN(t)) - if err != nil { - t.Skipf("postgres unavailable (%v) - skipping", err) - } - defer d.Close() - store, intentID, cleanup := newRunningInterventionIntent(t, d, "settlement retry") - defer cleanup() - - const requestID = "intervention-settle-1" - const message = "立刻停止当前动作并检查权限边界" - reserved, err := store.ReserveIntentIntervention(intentID, requestID, message) - if err != nil { - t.Fatal(err) - } - if _, _, err := store.AcceptIntentIntervention(intentID, requestID, message); !errors.Is(err, ErrIntentInterventionConflict) { - t.Fatalf("accept before worker settlement error=%v, want conflict", err) - } - pending, err := store.IntentInterventionByRequest(intentID, requestID) - if err != nil || pending == nil || pending.Status != IntentInterventionPending || pending.ActivityID != reserved.ActivityID { - t.Fatalf("failed acceptance lost reservation: pending=%+v err=%v", pending, err) - } - if changed, err := store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed { - t.Fatalf("settle intent: changed=%v err=%v", changed, err) - } - activity, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message) - if err != nil || duplicate || activity.ID <= reserved.ActivityID { - t.Fatalf("retry acceptance: activity=%+v duplicate=%v err=%v", activity, duplicate, err) - } -} diff --git a/server/engine.go b/server/engine.go index 165972d..b0b0a56 100644 --- a/server/engine.go +++ b/server/engine.go @@ -75,7 +75,6 @@ const ( ) var errWorkControlConflict = errors.New("work control conflict") -var errWorkInterventionConflict = errors.New("work intervention conflict") // retryableWorkerModelError excludes errors already handled by the task router. // In particular, a quota error after partial streaming advances the task cursor @@ -131,17 +130,6 @@ type Engine struct { // runWorkerStep has stopped writing and committed its final state. workMu sync.Mutex work map[int64]*workExecution - // interventionLocks serializes reservation, cancellation and message hand-off - // per intent. PostgreSQL separately serializes the durable transaction steps. - interventionLocks sync.Map // intentID -> *sync.Mutex - - // directedClaims keeps intents with a new human message ahead of the normal - // frontier without mutating their persisted priority. runWorkerStep installs - // the entry before releasing ControlWork, so a single worker slot cannot grab - // unrelated work in the paused->open hand-off gap. - directedMu sync.Mutex - directedClaims map[string][]int64 - directedReady map[int64]bool // steerBox queues planner course-corrections for a running work (keyed by intent // id). The worker's PreToolUse hook drains it before its next tool call and hands @@ -174,14 +162,7 @@ type taskRuntime struct { type workExecution struct { cancel context.CancelCauseFunc done chan error - action string // user action: pause | cancel | intervene -} - -type WorkInterventionResult struct { - RequestID string - ActivityID int64 - State string - Duplicate bool + action string // user action: pause | cancel } // nextPlannerRound returns the next planner round number for a task (1-based). @@ -304,12 +285,6 @@ func (e *Engine) StopTask(taskID string) { e.stamped.Delete(taskID) e.inflight.Delete(taskID) e.coordStarted.Delete(taskID) - e.directedMu.Lock() - for _, intentID := range e.directedClaims[taskID] { - delete(e.directedReady, intentID) - } - delete(e.directedClaims, taskID) - e.directedMu.Unlock() e.deleteMu.Lock() e.deleting.Delete(taskID) e.deleteMu.Unlock() @@ -415,7 +390,7 @@ func NewEngine(m *Manager) *Engine { return &Engine{m: m, debounce: 800 * time.Millisecond, bc: NewBroadcaster(), execCancel: map[string]context.CancelCauseFunc{}, execCtx: map[string]context.Context{}, work: map[int64]*workExecution{}, steerBox: map[int64][]string{}, - directedClaims: map[string][]int64{}, directedReady: map[int64]bool{}, runtimes: map[string]*taskRuntime{}} + runtimes: map[string]*taskRuntime{}} } // registerWork records the cancel for the work currently running intentID. @@ -450,7 +425,7 @@ func (e *Engine) detachWork(intentID int64) (action string, complete func(error) // worker has fully stopped writing. Cancellation cleanup is performed by the API // handler after this returns; pause state is committed by runWorkerStep itself. func (e *Engine) ControlWork(ctx context.Context, intentID int64, action string) error { - if action != "pause" && action != "cancel" && action != "intervene" { + if action != "pause" && action != "cancel" { return fmt.Errorf("unsupported work action %q", action) } if ctx == nil { @@ -469,11 +444,8 @@ func (e *Engine) ControlWork(ctx context.Context, intentID int64, action string) run.action = action done := run.done cause := error(agent.AbortWorkPausedByUser) - switch action { - case "cancel": + if action == "cancel" { cause = agent.AbortWorkCancelledByUser - case "intervene": - cause = agent.AbortWorkIntervenedByUser } run.cancel(cause) e.workMu.Unlock() @@ -492,282 +464,6 @@ func (e *Engine) ControlWork(ctx context.Context, intentID int64, action string) } } -func (e *Engine) lockWorkIntervention(intentID int64) func() { - value, _ := e.interventionLocks.LoadOrStore(intentID, &sync.Mutex{}) - mu := value.(*sync.Mutex) - mu.Lock() - return mu.Unlock -} - -func (e *Engine) validateWorkInterventionTask(t *Task, allowAdmissionPause bool) error { - if t == nil { - return fmt.Errorf("%w: task not found", errWorkInterventionConflict) - } - lifecycle := t.lifecycleSnapshot() - switch { - case e.IsDeleting(t.ID): - return fmt.Errorf("%w: 任务正在删除，无法向 Worker 发送消息", errWorkInterventionConflict) - case lifecycle.Paused || (!allowAdmissionPause && e.IsPaused(t.ID)): - return fmt.Errorf("%w: 任务已暂停，无法向 Worker 发送消息", errWorkInterventionConflict) - case lifecycle.Queued: - return fmt.Errorf("%w: 排队中的任务无法向 Worker 发送消息", errWorkInterventionConflict) - case isTerminalStatus(lifecycle.Status): - return fmt.Errorf("%w: 终态任务无法向 Worker 发送消息", errWorkInterventionConflict) - case e.isSettling(t.ID): - return fmt.Errorf("%w: 任务正在收尾，无法向 Worker 发送消息", errWorkInterventionConflict) - default: - return nil - } -} - -func (e *Engine) abandonWorkIntervention(t *Task, intentID int64, requestID string, reason error) { - e.removeDirectedClaim(t.ID, intentID) - target := "open" - lifecycle := t.lifecycleSnapshot() - if e.IsDeleting(t.ID) || isTerminalStatus(lifecycle.Status) { - target = "stopped" - } else if e.isSettling(t.ID) { - target = "exhausted" - } - if changed, err := t.Store.CompareAndSetIntentState(intentID, "paused", target); err != nil { - log.Printf("[worker intervention] task %s 意图 #%d 补偿 paused->%s 失败: %v", t.ID, intentID, target, err) - } else if changed && target == "open" { - t.NotifyWorker() - } - if err := t.Store.RejectIntentIntervention(intentID, requestID, reason.Error()); err != nil { - log.Printf("[worker intervention] task %s 意图 #%d 标记请求 %s rejected 失败: %v", t.ID, intentID, requestID, err) - } -} - -// InterveneWork performs the complete Worker message protocol. Once the request -// has a durable pending reservation it no longer depends on the HTTP request -// context: the caller supplies a server-owned bounded context so disconnects do -// not strand the intent. The accepted DB transition is running -> paused -> -// open, with the visible user message and reopen committed atomically. -func (e *Engine) InterveneWork(ctx context.Context, t *Task, intentID int64, requestID, message string) (WorkInterventionResult, error) { - return e.interveneWork(ctx, t, intentID, requestID, message, false) -} - -func (e *Engine) interveneWork(ctx context.Context, t *Task, intentID int64, requestID, message string, allowAdmissionPause bool) (WorkInterventionResult, error) { - out := WorkInterventionResult{RequestID: requestID} - if t == nil || t.Store == nil { - return out, fmt.Errorf("%w: task not found", errWorkInterventionConflict) - } - if ctx == nil { - ctx = context.Background() - } - unlock := e.lockWorkIntervention(intentID) - defer unlock() - - existing, err := t.Store.IntentInterventionByRequest(intentID, requestID) - if err != nil { - return out, err - } - if existing != nil { - if existing.Message != message { - return out, fmt.Errorf("%w: request_id 已用于不同的 Worker 消息", errWorkInterventionConflict) - } - if existing.Status == db.IntentInterventionAccepted { - out.ActivityID, out.State, out.Duplicate = existing.ActivityID, existing.IntentState, true - if out.State == "open" && e.validateWorkInterventionTask(t, allowAdmissionPause) == nil { - e.queueDirectedClaim(t.ID, intentID) - e.readyDirectedClaim(intentID) - t.NotifyWorker() - } - return out, nil - } - if existing.Status != db.IntentInterventionPending { - return out, fmt.Errorf("%w: Worker 消息请求状态为 %s", errWorkInterventionConflict, existing.Status) - } - } - if err := e.validateWorkInterventionTask(t, allowAdmissionPause); err != nil { - lifecycle := t.lifecycleSnapshot() - if existing != nil && (e.IsDeleting(t.ID) || e.isSettling(t.ID) || isTerminalStatus(lifecycle.Status)) { - e.abandonWorkIntervention(t, intentID, requestID, err) - } - return out, err - } - _, worker := e.snapshotFor(t) - if worker == nil { - return out, fmt.Errorf("%w: worker 尚未就绪", errWorkInterventionConflict) - } - if existing == nil { - existing, err = t.Store.ReserveIntentIntervention(intentID, requestID, message) - if err != nil { - if errors.Is(err, db.ErrIntentInterventionConflict) { - return out, fmt.Errorf("%w: %v", errWorkInterventionConflict, err) - } - return out, err - } - } - out.ActivityID = existing.ActivityID - - // A pending reservation can be recovered after a process restart. An open - // row is fenced back to paused before publishing the user turn; a running row - // is stopped through the normal live-work protocol so no late writer races us. - for { - node, getErr := t.Store.GetNode(intentID) - if getErr != nil { - return out, getErr - } - if node == nil { - return out, fmt.Errorf("%w: intent not found", errWorkInterventionConflict) - } - switch node.State { - case "running": - controlErr := e.ControlWork(ctx, intentID, "intervene") - if controlErr != nil { - // The per-work 30s wait may expire before the server-owned operation - // context. Keep waiting for the named cancellation to settle; timeout - // release marks the run no-handoff so a late exit cannot install a - // permanent directed-claim barrier behind our back. A freshly claimed - // intent can also be running just before registerWork; retrying closes - // that small claim-to-registration window. - for { - settled, readErr := t.Store.GetNode(intentID) - if readErr != nil { - return out, readErr - } - if settled != nil && settled.State == "paused" { - break - } - if ctx.Err() != nil || settled == nil || settled.State != "running" { - e.removeDirectedClaim(t.ID, intentID) - return out, fmt.Errorf("%w: %v", errWorkInterventionConflict, controlErr) - } - if sleepCtx(ctx, 50*time.Millisecond) { - e.removeDirectedClaim(t.ID, intentID) - return out, fmt.Errorf("%w: %v", errWorkInterventionConflict, controlErr) - } - if errors.Is(controlErr, errWorkControlConflict) { - controlErr = e.ControlWork(ctx, intentID, "intervene") - } - } - } - case "open": - changed, setErr := t.Store.CompareAndSetIntentState(intentID, "open", "paused") - if setErr != nil { - return out, setErr - } - if !changed { - continue - } - e.queueDirectedClaim(t.ID, intentID) - case "paused": - e.queueDirectedClaim(t.ID, intentID) - default: - e.abandonWorkIntervention(t, intentID, requestID, - fmt.Errorf("意图已进入 %s，无法发送新消息", node.State)) - return out, fmt.Errorf("%w: 意图已进入 %s，无法发送新消息", errWorkInterventionConflict, node.State) - } - break - } - e.queueDirectedClaim(t.ID, intentID) - - // Close lifecycle races that began while cancellation was settling. A task - // pause or queue preserves the pending message and paused intent for a later - // admission; terminal, settling and deleting transitions close the request. - if err := e.validateWorkInterventionTask(t, allowAdmissionPause); err != nil { - // The user message has not entered the transcript yet. Preserve the - // reservation and paused intent so the same request can be retried after a - // transient task pause; do not reopen it behind a terminal transition. - e.removeDirectedClaim(t.ID, intentID) - lifecycle := t.lifecycleSnapshot() - if e.IsDeleting(t.ID) || e.isSettling(t.ID) || isTerminalStatus(lifecycle.Status) { - e.abandonWorkIntervention(t, intentID, requestID, err) - } - return out, err - } - activity, duplicate, err := t.Store.AcceptIntentIntervention(intentID, requestID, message) - if err != nil { - // Keep this intent paused so a retry of the same request can safely finish; - // remove only the task-wide claim barrier so unrelated workers keep moving. - e.removeDirectedClaim(t.ID, intentID) - return out, err - } - if !duplicate { - e.publishStoredActivity(t, activity) - } - e.readyDirectedClaim(intentID) - t.NotifyWorker() - node, err := t.Store.GetNode(intentID) - if err != nil { - return out, err - } - out.Duplicate = duplicate - if node != nil { - out.State = node.State - } - return out, nil -} - -// RecoverWorkerMessages reconstructs unfinished Worker chat handoffs before -// ordinary frontier claiming. Pending messages are fenced in paused until an -// agent runtime is available; accepted messages regain their directed priority -// and are consumed by ExecuteWithMessage on the next claim. -func (e *Engine) RecoverWorkerMessages(ctx context.Context, t *Task) error { - return e.recoverWorkerMessages(ctx, t, false) -} - -// recoverWorkerMessagesForAdmission runs while the scheduler still owns the -// Engine pause barrier but has already committed a runnable task lifecycle. It -// restores directed hand-offs before Resume wakes any worker, so ordinary -// frontier work cannot overtake the human message. -func (e *Engine) recoverWorkerMessagesForAdmission(ctx context.Context, t *Task) error { - return e.recoverWorkerMessages(ctx, t, true) -} - -func (e *Engine) recoverWorkerMessages(ctx context.Context, t *Task, allowAdmissionPause bool) error { - if t == nil || t.Store == nil { - return nil - } - items, err := t.Store.RecoverableIntentInterventions() - if err != nil { - return err - } - lifecycle := t.lifecycleSnapshot() - runnable := !e.IsDeleting(t.ID) && !e.isSettling(t.ID) && !isTerminalStatus(lifecycle.Status) && - !lifecycle.Paused && !lifecycle.Queued && (allowAdmissionPause || !e.IsPaused(t.ID)) - if !runnable { - for _, item := range items { - e.removeDirectedClaim(t.ID, item.IntentID) - } - return nil - } - for _, item := range items { - switch item.Status { - case db.IntentInterventionPending: - if item.IntentState == "open" { - if changed, setErr := t.Store.CompareAndSetIntentState(item.IntentID, "open", "paused"); setErr != nil { - return setErr - } else if changed { - item.IntentState = "paused" - } - } - if _, worker := e.snapshotFor(t); worker == nil { - continue - } - if _, recoverErr := e.interveneWork(ctx, t, item.IntentID, item.RequestID, item.Message, allowAdmissionPause); recoverErr != nil { - log.Printf("[worker chat] task %s 恢复意图 #%d 待处理消息失败: %v", t.ID, item.IntentID, recoverErr) - } - case db.IntentInterventionAccepted: - if item.IntentState == "paused" { - if changed, setErr := t.Store.CompareAndSetIntentState(item.IntentID, "paused", "open"); setErr != nil { - return setErr - } else if changed { - item.IntentState = "open" - } - } - if item.IntentState == "open" { - e.queueDirectedClaim(t.ID, item.IntentID) - e.readyDirectedClaim(item.IntentID) - t.NotifyWorker() - } - } - } - return nil -} - // releaseWorkControl drops only this caller's reservation after its wait is // cancelled. The work context stays cancelled; runWorkerStep recognizes the // named cancellation cause and settles the intent into the recoverable paused @@ -775,11 +471,7 @@ func (e *Engine) recoverWorkerMessages(ctx context.Context, t *Task, allowAdmiss func (e *Engine) releaseWorkControl(intentID int64, run *workExecution, action string) { e.workMu.Lock() if current := e.work[intentID]; current == run && current.action == action { - if action == "intervene" { - current.action = "intervene_no_handoff" - } else { - current.action = "" - } + current.action = "" } e.workMu.Unlock() } @@ -918,19 +610,6 @@ func (e *Engine) emitActivity(t *Task, r db.Activity) db.Activity { return r } -// publishStoredActivity broadcasts an activity inserted by a larger DB -// transaction, avoiding a second INSERT through emitActivity. -func (e *Engine) publishStoredActivity(t *Task, r db.Activity) { - if t == nil || r.ID <= 0 { - return - } - if r.CreatedAt.IsZero() { - r.CreatedAt = time.Now() - } - e.bc.Publish(t.ID, r) - e.touch(t.ID) -} - // appendActivity persists one activity row, retrying briefly on write failure. // Concurrent planner + worker inserts into the same exploration's activity log // occasionally fail; a couple of quick retries recover most. Crucially, every @@ -1197,13 +876,13 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { } _, worker := e.snapshotFor(t) if worker == nil { - if waitWorker(ctx, t, 1500*time.Millisecond) { + if sleepCtx(ctx, 1500*time.Millisecond) { return } continue } if e.IsPaused(t.ID) { - if waitWorker(ctx, t, 1000*time.Millisecond) { + if sleepCtx(ctx, 1000*time.Millisecond) { return } continue // user-paused: don't claim/execute intents @@ -1212,13 +891,13 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { return } if e.isSettling(t.ID) { - if waitWorker(ctx, t, 1000*time.Millisecond) { + if sleepCtx(ctx, 1000*time.Millisecond) { return } continue // 任务超时收尾中:不再领新意图(在跑的自行收尾,协调器等其 drain) } if isTerminalStatus(e.m.TaskStatus(t.ID)) { - if waitWorker(ctx, t, 1000*time.Millisecond) { + if sleepCtx(ctx, 1000*time.Millisecond) { return } continue // 任务已终态(done/failed/timeout):停止领取遗留意图,别在完成后空跑 frontier @@ -1228,21 +907,34 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) { } claimed := e.runWorkerStep(ctx, t, name, worker) e.decInflight(t.ID) - if !claimed && waitWorker(ctx, t, 800*time.Millisecond) { + if !claimed && sleepCtx(ctx, 800*time.Millisecond) { return } } } -// 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) } @@ -1267,31 +959,16 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker workCtx = intercept.WithTaskContext(workCtx, t.ID, fmt.Sprintf("%s · #%d", name, iid), taskEmit) hooks := steerHooks{inner: t.Guard.Hooks(), drain: func() (string, bool) { return e.drainSteer(iid) }} wTaskID, _ := strconv.ParseInt(t.ID, 10, 64) - chatRequestID, chatMessage, hasChatMessage, chatErr := t.Store.PendingIntentInterventionMessage(intent.ID) - if chatErr != nil { - log.Printf("[worker %s] task %s 读取意图 #%d 待处理人工对话失败: %v", name, t.ID, intent.ID, chatErr) - if err := transitionIntentState(t.Store, intent.ID, "running", "open"); err != nil { - log.Printf("[worker %s] task %s 意图 #%d 人工对话读取失败后回退失败: %v", name, t.ID, intent.ID, err) - } - _, completeWork := e.detachWork(intent.ID) - completeWork(chatErr) - return true - } e.BeginLLMCall(t.ID) 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, chatRequestID, chatMessage) + 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) - if hasChatMessage { - if markErr := t.Store.MarkIntentInterventionHandoffClaimed(intent.ID); markErr != nil { - log.Printf("[worker %s] task %s 标记意图 #%d 人工对话已消费失败: %v", name, t.ID, intent.ID, markErr) - } - } // model_error 收场 → 额外重跑几次（退避后再试）。仅在意图仍属本 work、任务 // 未暂停/未终止/未取消【且未进入收尾】时重试；否则让位给对应分支处理(收尾期不 // 再重试,避免退避挤占其他 worker 的优雅收尾窗口)。 @@ -1305,7 +982,7 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker } e.BeginLLMCall(t.ID) if hasChatMessage { - reason, wrote, err = worker.ExecuteWithMessage(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding, chatRequestID, chatMessage) + 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) } @@ -1326,8 +1003,6 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker switch { case errors.Is(workCause, agent.AbortWorkPausedByUser): action = "pause" - case errors.Is(workCause, agent.AbortWorkIntervenedByUser): - action = "intervene" case errors.Is(workCause, agent.AbortWorkCancelledByUser): action = "cancel" } @@ -1344,22 +1019,6 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker e.touch(t.ID) return true } - if action == "intervene" || action == "intervene_no_handoff" { - controlErr = transitionIntentState(t.Store, intent.ID, "running", "paused") - if controlErr != nil { - log.Printf("[worker %s] task %s 意图 #%d 对话暂停落库失败: %v", name, t.ID, intent.ID, controlErr) - return true - } - // Install the directed barrier before completeWork releases ControlWork. - // With one worker slot this prevents its next loop from taking unrelated - // frontier work during the message hand-off. - if action == "intervene" { - e.queueDirectedClaim(t.ID, intent.ID) - } - log.Printf("[worker %s] task %s 意图 #%d 已暂停，等待新的对话意图", name, t.ID, intent.ID) - e.touch(t.ID) - return true - } if action == "cancel" { // Park the stopped run in paused before handing cleanup to the API. If the // request disconnects after cancellation, the intent remains recoverable and @@ -1441,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) { @@ -1464,114 +1178,7 @@ func sleepCtx(ctx context.Context, d time.Duration) (done bool) { } } -func waitWorker(ctx context.Context, t *Task, d time.Duration) (done bool) { - timer := time.NewTimer(d) - defer timer.Stop() - var wake <-chan struct{} - if t != nil { - wake = t.workerWake - } - select { - case <-ctx.Done(): - return true - case <-wake: - return false - case <-timer.C: - return false - } -} - -func (e *Engine) queueDirectedClaim(taskID string, intentID int64) { - if taskID == "" || intentID <= 0 { - return - } - e.directedMu.Lock() - defer e.directedMu.Unlock() - for _, id := range e.directedClaims[taskID] { - if id == intentID { - return - } - } - e.directedClaims[taskID] = append(e.directedClaims[taskID], intentID) - e.directedReady[intentID] = false -} - -func (e *Engine) readyDirectedClaim(intentID int64) { - e.directedMu.Lock() - if _, exists := e.directedReady[intentID]; exists { - e.directedReady[intentID] = true - } - e.directedMu.Unlock() -} - -func (e *Engine) removeDirectedClaim(taskID string, intentID int64) { - e.directedMu.Lock() - defer e.directedMu.Unlock() - items := e.directedClaims[taskID] - for i, id := range items { - if id != intentID { - continue - } - items = append(items[:i], items[i+1:]...) - delete(e.directedReady, intentID) - if len(items) == 0 { - delete(e.directedClaims, taskID) - } else { - e.directedClaims[taskID] = items - } - return - } -} - -// claimDirected tries every pending human-message hand-off before ordinary work -// frontier. blocked=true means at least one hand-off is still paused/running; -// callers must not let a just-released single worker slot overtake it. -func (e *Engine) claimDirected(t *Task, name string) (intent *db.Node, blocked bool) { - e.directedMu.Lock() - ids := append([]int64(nil), e.directedClaims[t.ID]...) - ready := make(map[int64]bool, len(ids)) - for _, id := range ids { - ready[id] = e.directedReady[id] - } - e.directedMu.Unlock() - for _, id := range ids { - node, err := t.Store.GetNode(id) - if err != nil { - return nil, true - } - if node == nil { - e.removeDirectedClaim(t.ID, id) - continue - } - switch node.State { - case "open": - if !ready[id] { - blocked = true - continue - } - ok, err := t.Store.ClaimIntent(id, name) - if err != nil { - return nil, true - } - if ok { - e.removeDirectedClaim(t.ID, id) - node.State, node.Owner = "running", name - return node, false - } - blocked = true - case "paused", "running": - blocked = true - default: - e.removeDirectedClaim(t.ID, id) - } - } - return nil, blocked -} - func (e *Engine) claimNext(t *Task, name string) *db.Node { - if intent, blocked := e.claimDirected(t, name); intent != nil || blocked { - return intent - } fr, _ := t.Store.Frontier(20) for _, in := range fr { if ok, _ := t.Store.ClaimIntent(in.ID, name); ok { diff --git a/server/engine_cancelcause_test.go b/server/engine_cancelcause_test.go index 63c7324..d1f4f7a 100644 --- a/server/engine_cancelcause_test.go +++ b/server/engine_cancelcause_test.go @@ -45,7 +45,6 @@ func TestControlWorkCarriesUserActionCause(t *testing.T) { }{ {action: "pause", code: "work_paused_by_user"}, {action: "cancel", code: "work_cancelled_by_user"}, - {action: "intervene", code: "work_intervened_by_user"}, } { t.Run(tc.action, func(t *testing.T) { e := NewEngine(nil) diff --git a/server/goals.go b/server/goals.go index 05f2c7e..c8e6d98 100644 --- a/server/goals.go +++ b/server/goals.go @@ -225,13 +225,6 @@ func (s *Server) hasReadyQueuedTask(excludeID string) bool { } func (s *Server) startAdmittedTask(t *Task, mode string) { - recoverMessages := s.engine.RecoverWorkerMessages - if s.engine.IsPaused(t.ID) { - recoverMessages = s.engine.recoverWorkerMessagesForAdmission - } - if err := recoverMessages(s.ctx, t); err != nil { - log.Printf("[worker chat] task %s 恢复待处理人工消息失败: %v", t.ID, err) - } if mode == "bootstrap" { if !s.engine.beginTaskOperation(t.ID) { return diff --git a/server/intent_intervention.go b/server/intent_intervention.go index 6df99fa..3e3655c 100644 --- a/server/intent_intervention.go +++ b/server/intent_intervention.go @@ -1,13 +1,11 @@ package server import ( - "context" "encoding/json" "errors" "net/http" "strconv" "strings" - "time" "github.com/Autumn-27/artex/db" ) @@ -28,9 +26,15 @@ func validWorkerMessageRequestID(id string) bool { return true } -// sendWorkerMessage pauses a running Worker when necessary, records one user -// turn for its existing conversation, and immediately schedules that same -// intent to continue. A disconnected browser cannot strand an accepted message. +// 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 { @@ -71,11 +75,28 @@ func (s *Server) sendWorkerMessage(w http.ResponseWriter, r *http.Request) { return } - if !s.engine.beginTaskOperation(t.ID) { + // 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 } - defer s.engine.decInflight(t.ID) + 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 { @@ -94,19 +115,17 @@ func (s *Server) sendWorkerMessage(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusConflict, "node is not an intent") return } - base := s.ctx - if base == nil { - base = context.Background() + if node.State != "paused" { + writeErr(w, http.StatusConflict, "仅已暂停的 Worker 可以发送消息，请先暂停") + return } - opCtx, cancel := context.WithTimeout(base, workControlWaitTimeout+15*time.Second) - defer cancel() - result, err := s.engine.InterveneWork(opCtx, t, iid, requestID, message) - if err != nil { + + // 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, context.DeadlineExceeded), errors.Is(err, context.Canceled): - writeErr(w, http.StatusGatewayTimeout, err.Error()) - case errors.Is(err, errWorkInterventionConflict), errors.Is(err, errWorkControlConflict), - errors.Is(err, db.ErrIntentInterventionConflict), errors.Is(err, db.ErrIntentStateConflict): + case errors.Is(err, db.ErrIntentStateConflict): writeErr(w, http.StatusConflict, err.Error()) default: writeErr(w, http.StatusInternalServerError, err.Error()) @@ -115,11 +134,9 @@ func (s *Server) sendWorkerMessage(w http.ResponseWriter, r *http.Request) { } writeJSON(w, http.StatusOK, map[string]any{ - "id": iid, - "state": result.State, - "accepted": true, - "activity_seq": result.ActivityID, - "request_id": result.RequestID, - "duplicate": result.Duplicate, + "id": iid, + "state": "running", + "accepted": true, + "request_id": requestID, }) } diff --git a/server/intent_intervention_protocol_test.go b/server/intent_intervention_protocol_test.go deleted file mode 100644 index 8347eca..0000000 --- a/server/intent_intervention_protocol_test.go +++ /dev/null @@ -1,378 +0,0 @@ -package server - -import ( - "bytes" - "context" - "encoding/json" - "errors" - "iter" - "net/http" - "net/http/httptest" - "strconv" - "strings" - "testing" - "time" - - "github.com/Autumn-27/artex/agent" - "github.com/Autumn-27/artex/db" - "github.com/Autumn-27/norma/llm" - "github.com/Autumn-27/norma/transcript" -) - -type workerMessageTestProvider struct{} - -func (workerMessageTestProvider) Stream(_ context.Context, _ llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { - return func(yield func(llm.StreamEvent, error) bool) { - yield(llm.StreamEvent{Type: llm.SETextDelta, Text: "已按新的对话意图继续执行"}, nil) - yield(llm.StreamEvent{Type: llm.SEMessageDelta, StopReason: "end_turn"}, nil) - yield(llm.StreamEvent{Type: llm.SEMessageStop}, nil) - } -} - -func (p workerMessageTestProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { - acc := llm.NewAccumulator() - for event, err := range p.Stream(ctx, req) { - if err != nil { - return llm.Message{}, "", llm.Usage{}, err - } - acc.Add(event) - } - return acc.Message(), acc.StopReason, acc.Usage, nil -} - -type interventionProtocolFixture struct { - m *Manager - task *Task - engine *Engine - worker *agent.Worker - tx *transcript.Store - intentID int64 -} - -func newInterventionProtocolFixture(t *testing.T, targetPriority int) *interventionProtocolFixture { - t.Helper() - m, err := NewManager(t.TempDir(), "") - if err != nil { - t.Skipf("postgres unavailable (%v) - skipping", err) - } - task, err := m.CreateTask("worker message protocol", "verify immediate conversation handoff", nil, 0, 0) - if err != nil { - m.Close() - t.Fatal(err) - } - intentID, err := task.Store.AddIntent(map[string]any{"summary": "target worker intent"}, targetPriority, nil, "planner") - if err != nil { - m.Close() - t.Fatal(err) - } - if claimed, err := task.Store.ClaimIntent(intentID, "work#1"); err != nil || !claimed { - m.Close() - t.Fatalf("claim target intent: claimed=%v err=%v", claimed, err) - } - tx := transcript.NewStore(m.dir + "/transcripts") - worker := agent.NewWorker(workerMessageTestProvider{}, "test-model", m.dir, tx, 0, 1) - engine := NewEngine(m) - engine.UseLLM(nil, worker) - fixture := &interventionProtocolFixture{m: m, task: task, engine: engine, worker: worker, tx: tx, intentID: intentID} - t.Cleanup(func() { - taskID, _ := strconv.ParseInt(task.ID, 10, 64) - _ = m.pg.DeleteTask(taskID) - _ = m.Close() - }) - return fixture -} - -// settleRegisteredIntervention emulates the ordering inside runWorkerStep after -// a cancellation-aware provider returns: detach the live work handle, commit the -// running->paused final state, install the directed hand-off, then release the -// controller waiting on workExecution.done. -func settleRegisteredIntervention(t *testing.T, f *interventionProtocolFixture) string { - t.Helper() - action, complete := f.engine.detachWork(f.intentID) - if action != "intervene" { - t.Fatalf("settled action=%q, want intervene", action) - } - if changed, err := f.task.Store.CompareAndSetIntentState(f.intentID, "running", "paused"); err != nil || !changed { - t.Fatalf("settle running->paused: changed=%v err=%v", changed, err) - } - f.engine.queueDirectedClaim(f.task.ID, f.intentID) - complete(nil) - return action -} - -func countTranscriptWorkerMessages(t *testing.T, tx *transcript.Store, sessionID, requestID string) (int, []transcript.Record) { - t.Helper() - records, err := tx.Load(tx.MainPath(sessionID)) - if err != nil { - t.Fatal(err) - } - marker := "ARTEX_WORKER_CHAT:" + requestID - count := 0 - for _, record := range records { - if record.Message != nil && strings.Contains(record.Message.Text(), marker) { - count++ - } - } - return count, records -} - -func TestWorkerMessageCancelsBeforeAcceptAndContinuesSameSessionFirst(t *testing.T) { - f := newInterventionProtocolFixture(t, 1) - competitorID, err := f.task.Store.AddIntent(map[string]any{"summary": "higher priority competitor"}, 100, nil, "planner") - if err != nil { - t.Fatal(err) - } - - // Pre-seed the same stable session so the message must continue its UUID - // chain rather than creating a worker-slot-specific conversation. - sessionID := agent.WorkerSessionID(f.task.ExpID, f.intentID) - seed := f.tx.NewWriter(sessionID, "") - seed.RecordMessage(llm.UserText("original intent turn"), llm.Usage{}) - if err := seed.Err(); err != nil { - t.Fatal(err) - } - before, err := f.tx.Load(f.tx.MainPath(sessionID)) - if err != nil || len(before) != 1 { - t.Fatalf("seed transcript records=%d err=%v", len(before), err) - } - - workCtx, cancelWork := context.WithCancelCause(context.Background()) - f.engine.registerWork(f.intentID, cancelWork) - live, unsubscribe := f.engine.Broadcaster().Subscribe(f.task.ID) - defer unsubscribe() - - const requestID = "worker-message-engine-1" - const message = "停止主动枚举，立即验证登录越权" - type outcome struct { - result WorkInterventionResult - err error - } - done := make(chan outcome, 1) - go func() { - result, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, message) - done <- outcome{result: result, err: err} - }() - - select { - case <-workCtx.Done(): - if code, _, _, _ := agent.AbortReason(workCtx); code != "work_intervened_by_user" { - t.Fatalf("old execution cancel code=%q, want work_intervened_by_user", code) - } - case <-time.After(time.Second): - t.Fatal("Worker message did not immediately cancel the old execution") - } - select { - case early := <-done: - t.Fatalf("Worker message accepted before old worker settled: %+v", early) - case <-time.After(25 * time.Millisecond): - } - pending, err := f.task.Store.IntentInterventionByRequest(f.intentID, requestID) - if err != nil || pending == nil || pending.Status != db.IntentInterventionPending { - t.Fatalf("durable pending reservation=%+v err=%v", pending, err) - } - settleRegisteredIntervention(t, f) - - var accepted outcome - select { - case accepted = <-done: - case <-time.After(2 * time.Second): - t.Fatal("timed out waiting for accepted Worker message") - } - if accepted.err != nil || accepted.result.ActivityID <= pending.ActivityID || accepted.result.Duplicate || accepted.result.State != "open" { - t.Fatalf("accepted Worker message=%+v control=%+v err=%v", accepted.result, pending, accepted.err) - } - node, err := f.task.Store.GetNode(f.intentID) - if err != nil || node == nil || node.State != "open" { - t.Fatalf("reopened target=%+v err=%v", node, err) - } - - // The API publishes the visible user activity but never edits transcript files. - // agentcore appends the standard user turn only after the same Worker is claimed. - count, records := countTranscriptWorkerMessages(t, f.tx, sessionID, requestID) - if count != 0 || len(records) != 1 { - t.Fatalf("HTTP handler edited Worker transcript: count=%d records=%+v", count, records) - } - select { - case activity := <-live: - if activity.ID != accepted.result.ActivityID || activity.NodeID == nil || *activity.NodeID != f.intentID || - activity.Worker != "user" || activity.Kind != "user" || activity.Summary != message { - t.Fatalf("broadcast activity=%+v", activity) - } - case <-time.After(time.Second): - t.Fatal("accepted Worker message was not broadcast") - } - select { - case duplicate := <-live: - t.Fatalf("accepted Worker message broadcast twice: %+v", duplicate) - default: - } - - // The directed hand-off must outrank the normal frontier without changing the - // persisted priority. This catches the one-worker race where the released slot - // used to take an unrelated, higher-priority intent before the target reopened. - claimed := f.engine.claimNext(f.task, "work#2") - if claimed == nil || claimed.ID != f.intentID { - t.Fatalf("directed claim=%+v, want target %d ahead of competitor %d", claimed, f.intentID, competitorID) - } - competitor, err := f.task.Store.GetNode(competitorID) - if err != nil || competitor == nil || competitor.State != "open" { - t.Fatalf("competitor was claimed before Worker message target: node=%+v err=%v", competitor, err) - } - request, nextMessage, hasMessage, err := f.task.Store.PendingIntentInterventionMessage(f.intentID) - if err != nil || !hasMessage || request != requestID || nextMessage != message { - t.Fatalf("pending Worker message request=%q message=%q present=%v err=%v", request, nextMessage, hasMessage, err) - } - taskID, _ := strconv.ParseInt(f.task.ID, 10, 64) - if _, _, err := f.worker.ExecuteWithMessage(context.Background(), "work#2", taskID, nil, f.task.Store, - claimed, nil, nil, nil, nil, request, nextMessage); err != nil { - t.Fatalf("continue Worker conversation: %v", err) - } - if err := f.task.Store.MarkIntentInterventionHandoffClaimed(f.intentID); err != nil { - t.Fatal(err) - } - count, records = countTranscriptWorkerMessages(t, f.tx, sessionID, requestID) - if count != 1 || len(records) != 3 || records[1].ParentUUID != records[0].UUID || records[2].ParentUUID != records[1].UUID { - t.Fatalf("Worker message did not form one normal chained turn: count=%d records=%+v", count, records) - } - if records[1].Message == nil || records[1].Message.Role != llm.RoleUser || !strings.Contains(records[1].Message.Text(), message) { - t.Fatalf("Worker message was not recorded as the next user turn: %+v", records[1]) - } - if _, _, pendingAfterRun, err := f.task.Store.PendingIntentInterventionMessage(f.intentID); err != nil || pendingAfterRun { - t.Fatalf("Worker message handoff remains pending=%v err=%v", pendingAfterRun, err) - } - - retry, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, message) - if err != nil || !retry.Duplicate || retry.ActivityID != accepted.result.ActivityID { - t.Fatalf("idempotent engine retry=%+v err=%v", retry, err) - } - if _, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, "同 request_id 的另一条消息"); !errors.Is(err, errWorkInterventionConflict) { - t.Fatalf("same request_id different message error=%v, want conflict", err) - } - count, _ = countTranscriptWorkerMessages(t, f.tx, sessionID, requestID) - if count != 1 { - t.Fatalf("engine retries appended %d Worker messages, want 1", count) - } -} - -func TestSendWorkerMessageContinuesAfterRequestContextCancellation(t *testing.T) { - f := newInterventionProtocolFixture(t, 1) - workCtx, cancelWork := context.WithCancelCause(context.Background()) - f.engine.registerWork(f.intentID, cancelWork) - s := &Server{m: f.m, engine: f.engine, ctx: context.Background()} - - requestBody, _ := json.Marshal(map[string]string{ - "message": "中止当前动作，只验证权限边界", "request_id": "worker-message-disconnect-1", - }) - requestCtx, disconnect := context.WithCancel(context.Background()) - disconnect() - req := httptest.NewRequest(http.MethodPost, "/api/tasks/1/intents/1/messages", bytes.NewReader(requestBody)).WithContext(requestCtx) - req.SetPathValue("id", f.task.ID) - req.SetPathValue("iid", strconv.FormatInt(f.intentID, 10)) - recorder := httptest.NewRecorder() - done := make(chan struct{}) - go func() { - s.sendWorkerMessage(recorder, req) - close(done) - }() - - select { - case <-workCtx.Done(): - case <-time.After(time.Second): - t.Fatal("disconnected request did not continue to cancel the worker") - } - settleRegisteredIntervention(t, f) - select { - case <-done: - case <-time.After(2 * time.Second): - t.Fatal("server-owned Worker message did not finish after browser disconnect") - } - if recorder.Code != http.StatusOK { - t.Fatalf("disconnected Worker message status=%d body=%s", recorder.Code, recorder.Body.String()) - } - accepted, err := f.task.Store.IntentInterventionByRequest(f.intentID, "worker-message-disconnect-1") - if err != nil || accepted == nil || accepted.Status != db.IntentInterventionAccepted { - t.Fatalf("disconnected request acceptance=%+v err=%v", accepted, err) - } -} - -func TestWorkerMessageRejectsUnavailableTaskStatesWithoutReservation(t *testing.T) { - cases := []struct { - name string - block func(*interventionProtocolFixture) - }{ - {name: "paused", block: func(f *interventionProtocolFixture) { - f.task.updateLifecycle(func(state *taskLifecycleState) { state.Paused = true }) - f.engine.Pause(f.task.ID, agent.AbortPausedByUser) - }}, - {name: "queued", block: func(f *interventionProtocolFixture) { - f.task.updateLifecycle(func(state *taskLifecycleState) { state.Queued = true }) - }}, - {name: "terminal", block: func(f *interventionProtocolFixture) { - f.task.updateLifecycle(func(state *taskLifecycleState) { state.Status = "done" }) - }}, - {name: "settling", block: func(f *interventionProtocolFixture) { f.engine.markSettling(f.task.ID) }}, - {name: "deleting", block: func(f *interventionProtocolFixture) { f.engine.BeginDelete(f.task.ID) }}, - } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - f := newInterventionProtocolFixture(t, 1) - tc.block(f) - requestID := "worker-message-unavailable-" + tc.name - if _, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, "must not persist"); !errors.Is(err, errWorkInterventionConflict) { - t.Fatalf("unavailable state error=%v, want conflict", err) - } - reservation, err := f.task.Store.IntentInterventionByRequest(f.intentID, requestID) - if err != nil || reservation != nil { - t.Fatalf("unavailable state persisted reservation=%+v err=%v", reservation, err) - } - node, err := f.task.Store.GetNode(f.intentID) - if err != nil || node == nil || node.State != "running" { - t.Fatalf("unavailable state mutated intent=%+v err=%v", node, err) - } - count, _ := countTranscriptWorkerMessages(t, f.tx, agent.WorkerSessionID(f.task.ExpID, f.intentID), requestID) - if count != 0 { - t.Fatalf("unavailable state appended %d transcript interventions", count) - } - }) - } -} - -func TestWorkerWakeInterruptsLongIdlePoll(t *testing.T) { - task := &Task{workerWake: make(chan struct{}, 1)} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - started := make(chan struct{}) - returned := make(chan bool, 1) - go func() { - close(started) - returned <- waitWorker(ctx, task, 5*time.Second) - }() - <-started - start := time.Now() - task.NotifyWorker() - select { - case stopped := <-returned: - if stopped { - t.Fatal("worker wake was mistaken for context shutdown") - } - if elapsed := time.Since(start); elapsed > 250*time.Millisecond { - t.Fatalf("worker wake took %s, want immediate wake", elapsed) - } - case <-time.After(time.Second): - t.Fatal("worker wake waited for the idle polling interval") - } -} - -func TestSendWorkerMessageRejectsOversizedBodyAsEntityTooLarge(t *testing.T) { - f := newInterventionProtocolFixture(t, 1) - s := &Server{m: f.m, engine: f.engine, ctx: context.Background()} - body := `{"message":"` + strings.Repeat("x", maxWorkerMessageBytes) + `","request_id":"oversized"}` - req := httptest.NewRequest(http.MethodPost, "/api/tasks/1/intents/1/messages", strings.NewReader(body)) - req.SetPathValue("id", f.task.ID) - req.SetPathValue("iid", strconv.FormatInt(f.intentID, 10)) - recorder := httptest.NewRecorder() - s.sendWorkerMessage(recorder, req) - if recorder.Code != http.StatusRequestEntityTooLarge { - t.Fatalf("oversized Worker message status=%d body=%s", recorder.Code, recorder.Body.String()) - } -} diff --git a/server/manager.go b/server/manager.go index 6fca4dc..cb198c2 100644 --- a/server/manager.go +++ b/server/manager.go @@ -63,7 +63,6 @@ type Task struct { Store *pgdb.ExplorationStore `json:"-"` Guard *guard.Guard `json:"-"` notify chan struct{} - workerWake chan struct{} lifecycleMu sync.RWMutex llmMu sync.RWMutex @@ -738,7 +737,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds, CoverageEnabled: pt.CoverageEnabled, FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt), - Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1), workerWake: make(chan struct{}, 64), + Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1), } } @@ -1639,20 +1638,6 @@ func (t *Task) Notify() { } } -// NotifyWorker wakes one idle execution slot without also scheduling a Planner -// round. The channel is separate from notify because human intervention needs -// the reopened intent reclaimed immediately, while graph-change notifications -// are Planner-facing and debounced. A nil channel is tolerated by test fixtures. -func (t *Task) NotifyWorker() { - if t == nil || t.workerWake == nil { - return - } - select { - case t.workerWake <- struct{}{}: - default: - } -} - // NotifyDone is Notify plus a hint: a worker just finished intentID and that is // what triggered this wake-up. The planner reads the accumulated triggers next // round so it can spell out which intent finished (+ its output). Events pile up diff --git a/server/server.go b/server/server.go index b6a4182..f820b1e 100644 --- a/server/server.go +++ b/server/server.go @@ -249,9 +249,6 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi if n, _ := t.Store.ResetRunningIntents(); n > 0 { log.Printf("[engine] task %s 重置 %d 个残留 running 意图为 open", t.ID, n) } - if err := s.engine.RecoverWorkerMessages(ctx, t); err != nil { - log.Printf("[worker chat] task %s 恢复待处理人工消息失败: %v", t.ID, err) - } if lifecycle.Paused { s.engine.Pause(t.ID, agent.AbortPausedOnReload) } @@ -474,11 +471,6 @@ func (s *Server) applyLLM(cfg agent.Config) error { s.llmOn = true s.cfgMu.Unlock() s.invalidateTaskAgents() - for _, task := range s.m.List() { - if err := s.engine.RecoverWorkerMessages(s.ctx, task); err != nil { - log.Printf("[worker chat] task %s 在 LLM 就绪后恢复待处理人工消息失败: %v", task.ID, err) - } - } // wake the active task so a task created while idle starts exploring. if t := s.m.ActiveTask(); t != nil { 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 46ed163..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 @@ -1336,14 +1336,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { function sendWorkerChat() { const intentId = active.intent_id; const message = workerMessage.trim(); - if ( - !intentId || - active.inherited || - (active.status !== "paused" && active.status !== "pending") || - workerMessageSending || - !message - ) - return; + if (!intentId || active.inherited || active.status !== "paused" || workerMessageSending || !message) return; if (workerMessageCharCount(message) > MAX_WORKER_MESSAGE_CHARS) { toast.error(`消息不能超过 ${MAX_WORKER_MESSAGE_CHARS} 个字符`); return; @@ -1354,23 +1347,8 @@ export function SessionsTab({ taskId }: { taskId: string }) { api .sendWorkerMessage(taskId, intentId, message, requestId) .then((result) => { - const sentAt = new Date().toISOString(); - const activity: Activity = { - seq: result.activity_seq, - intent_id: intentId, - worker: "user", - ts: sentAt, - kind: "user", - summary: message, - detail: message, - }; - patchStore(`intent:${intentId}`, (state) => ({ - ...state, - items: mergeBySeq(state.items, [activity]), - lastTs: sentAt, - loaded: true, - unread: 0, - })); + // 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(""); @@ -1788,7 +1766,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { ) : active.role === "worker" && !active.inherited && - (active.status === "running" || active.status === "paused" || active.status === "pending") ? ( + (active.status === "running" || active.status === "paused") ? (
) : ( - MAX_WORKER_MESSAGE_CHARS - } - title="发送消息" - aria-label="发送消息" - > - {workerMessageSending ? : } - + <> + 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 ? : } + + )} diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 4516702..f2b5b3f 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -287,9 +287,8 @@ export const api = { sendWorkerMessage: (taskId: string, intentId: string, message: string, requestId: string) => post<{ id: number; - state: "open" | "running" | "paused"; + state: "running"; accepted: true; - activity_seq: number; request_id: string; }>(`/tasks/${taskId}/intents/${intentId}/messages`, { message, request_id: requestId }), taskLLMResolution: (id: string) => get(`/tasks/${id}/llm/resolution`), diff --git a/web/src/lib/mock/handler.ts b/web/src/lib/mock/handler.ts index 98b6c44..efa66f5 100644 --- a/web/src/lib/mock/handler.ts +++ b/web/src/lib/mock/handler.ts @@ -465,23 +465,23 @@ function sendMockWorkerMessage( } const intent = mockIntents.find((item) => item.id === id); if (!intent) return { ok: false, error: "意图不存在" }; - if (intent.inherited || (intent.state !== "running" && intent.state !== "paused")) { - return { ok: false, state: intent.state, error: "Worker 状态已变化" }; + 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 stops the active turn, persists a normal user message, - // and immediately reopens the same intent. Keep the mock's transition in sync, - // then emulate a Worker reclaiming it shortly afterward. - intent.state = "open"; + // 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: "open", + state: "running", activitySeq, }; mockWorkerMessages.set(normalizedRequestId, workerMessage); @@ -494,12 +494,6 @@ function sendMockWorkerMessage( summary: normalizedMessage, detail: normalizedMessage, } satisfies Activity); - setTimeout(() => { - if (intent.state === "open") { - intent.state = "running"; - workerMessage.state = "running"; - } - }, 250); return { ok: true, state: workerMessage.state, @@ -1008,9 +1002,8 @@ 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 ?? "open", + state: result.state ?? "running", accepted: true, - activity_seq: result.activitySeq ?? 0, request_id: result.requestId ?? "", }; }