Recover interrupted runs and commit terminal state once

Why: A hard kill left runs in pending, running, or waiting_hitl, a busy send stored a message before the 409, finishRun could mark a run completed without an assistant message, and a full SSE buffer dropped events without closing the stream.
Evidence: GOWORK=off go vet ./... and GOWORK=off go test -count=1 ./... passed, including store, runmgr, bus, and server recovery tests.
Impact: Migration v14 adds runs.instance_id and messages.request_id. Startup reconciliation and shutdown share one seal. Turn and workflow sends claim a slot before writing. Duplicate request ids return the existing message. SettleRun updates the run and appends the assistant message in one transaction. A full subscriber buffer closes the channel so SSE can replay from the cursor.
Refs: Dev feature/pi-durable-borrow
Co-authored-by: Cursor <cursoragent@cursor.com>

cmd/agent-system/main.go

@@ -41,13 +41,14 @@ func main() {
 	dsn := fs.String("dsn", "", "PostgreSQL 档 DSN（空 = 嵌入式 SQLite 档）")
 	artifacts := fs.String("artifacts", "", "工件对象目录（默认 <data>/artifacts；PG 档多实例时指向共享卷）")
 	auth := fs.String("auth", "auto", "API key 认证：auto（PG 档开、嵌入式关）| on | off")
+	instance := fs.String("instance", "", "本进程标识（空 = 主机名:监听地址；启动对账只收口这个实例的 run）")
 	fs.Parse(os.Args[1:])
 	cliSet := map[string]bool{}
 	fs.Visit(func(f *flag.Flag) { cliSet[f.Name] = true })
 
 	resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{
 		Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock,
-		StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth,
+		StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance,
 	}, cliSet)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, "agent-system:", err)
@@ -56,6 +57,7 @@ func main() {
 	if err := app.Run(app.Options{
 		Addr: resolved.Addr, DataDir: resolved.DataDir, PluginsDir: resolved.PluginsDir, ForceMock: resolved.Mock,
 		StorageDSN: resolved.StorageDSN, ArtifactsDir: resolved.ArtifactsDir, AuthMode: resolved.AuthMode,
+		InstanceID: resolved.Instance,
 	}); err != nil {
 		fmt.Fprintln(os.Stderr, "agent-system:", err)
 		os.Exit(1)
@@ -77,12 +79,13 @@ func runConfig(args []string) int {
 	dsn := fs.String("dsn", "", "")
 	artifacts := fs.String("artifacts", "", "")
 	auth := fs.String("auth", "auto", "")
+	instance := fs.String("instance", "", "")
 	fs.Parse(args[1:])
 	cliSet := map[string]bool{}
 	fs.Visit(func(f *flag.Flag) { cliSet[f.Name] = true })
 	resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{
 		Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock,
-		StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth,
+		StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance,
 	}, cliSet)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, "agent-system:", err)

internal/app/app.go

@@ -58,6 +58,7 @@ type Options struct {
 	StorageDSN   string // 空 = 嵌入式 SQLite 档；非空 = PostgreSQL 档（多用户/多实例）
 	ArtifactsDir string // PG 档工件对象目录（空 = <DataDir>/artifacts）
 	AuthMode     string // auto（默认：PG 档开、嵌入式关）| on | off
+	InstanceID   string // 空 = 主机名:监听地址；写入 runs.instance_id，启动对账只收口本实例
 }
 
 // Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。
@@ -187,6 +188,19 @@ func Run(opt Options) error {
 	subRunner := &runtime.SubLLMRunner{Prompts: promptSvc}
 	mockRunner := &runtime.MockRunner{}
 	mgr := runmgr.New(st, b, reg, profSvc, setSvc, convRunner, subRunner, mockRunner, &runtime.MockSubagent{})
+	instanceID := opt.InstanceID
+	if instanceID == "" {
+		host, err := os.Hostname()
+		if err != nil || host == "" {
+			host = "local"
+		}
+		instanceID = host + ":" + opt.Addr
+	}
+	st.SetInstanceID(instanceID)
+	mgr.SetInstanceID(instanceID)
+	if err := mgr.ReconcileInterrupted(); err != nil {
+		return fmt.Errorf("reconcile interrupted runs: %w", err)
+	}
 	mgr.SetSquadRunner(&runtime.SquadRunner{})
 
 	// 意图闸门：插件意图 overlay → agent-intent 注册表（未装载意图 = 行为回退）。

internal/bus/bus.go

@@ -1,5 +1,6 @@
 // Package bus 是 streamhub：按会话分发已持久化的事件（SSE 通道）。
-// 事件权威在 store（seq 游标）；bus 只做在线转发，溢出丢弃由客户端从游标重放补偿。
+// 事件权威在 store（seq 游标）；bus 只做在线转发。
+// 订阅缓冲满时关闭该订阅，客户端用 Last-Event-ID 从游标重放，不再静默丢事件。
 package bus
 
 import (
@@ -11,42 +12,64 @@ import (
 type subscriber struct {
 	ch     chan core.StreamEvent
 	filter string // conversationID；空 = 全部
+	closed bool
 }
 
 type Bus struct {
 	mu   sync.Mutex
 	subs map[int64]*subscriber
 	next int64
+	buf  int
 }
 
-func New() *Bus { return &Bus{subs: map[int64]*subscriber{}} }
+func New() *Bus { return NewWithBuffer(512) }
 
-// Subscribe 返回新订阅；close(unsubscribe) 注销。
+// NewWithBuffer 指定订阅缓冲长度。测试用小缓冲观察溢出关闭。
+func NewWithBuffer(n int) *Bus {
+	if n < 1 {
+		n = 1
+	}
+	return &Bus{subs: map[int64]*subscriber{}, buf: n}
+}
+
+// Subscribe 返回新订阅；close(unsubscribe) 注销。注销是幂等的。
 func (b *Bus) Subscribe(conversationID string) (<-chan core.StreamEvent, func()) {
 	b.mu.Lock()
 	id := b.next
 	b.next++
-	ch := make(chan core.StreamEvent, 512)
+	ch := make(chan core.StreamEvent, b.buf)
 	b.subs[id] = &subscriber{ch: ch, filter: conversationID}
 	b.mu.Unlock()
 	return ch, func() {
 		b.mu.Lock()
-		delete(b.subs, id)
+		b.dropLocked(id)
 		b.mu.Unlock()
 	}
 }
 
-// Publish 非阻塞广播；满则丢（客户端 Last-Event-ID 重放补齐）。
+// Publish 非阻塞广播。缓冲满的订阅被移除并关闭，避免静默丢事件。
 func (b *Bus) Publish(e core.StreamEvent) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	for _, s := range b.subs {
-		if s.filter != "" && s.filter != e.ConversationID {
+	for id, s := range b.subs {
+		if s.closed || (s.filter != "" && s.filter != e.ConversationID) {
 			continue
 		}
 		select {
 		case s.ch <- e:
 		default:
+			b.dropLocked(id)
 		}
 	}
 }
+
+// dropLocked 在 b.mu 下关闭并移除订阅。重复调用无效果。
+func (b *Bus) dropLocked(id int64) {
+	s, ok := b.subs[id]
+	if !ok || s.closed {
+		return
+	}
+	s.closed = true
+	delete(b.subs, id)
+	close(s.ch)
+}

internal/bus/bus_test.go

@@ -0,0 +1,37 @@
+package bus
+
+import (
+	"testing"
+
+	core "agent-system-core"
+)
+
+func TestPublishClosesOverflowedSubscriber(t *testing.T) {
+	b := NewWithBuffer(1)
+	ch, unsub := b.Subscribe("c1")
+	defer unsub()
+	b.Publish(core.StreamEvent{ConversationID: "c1", Seq: 1, Type: "a"})
+	b.Publish(core.StreamEvent{ConversationID: "c1", Seq: 2, Type: "b"})
+	n := 0
+	for {
+		e, ok := <-ch
+		if !ok {
+			break
+		}
+		n++
+		if e.Seq != 1 {
+			t.Fatalf("buffered event seq %d", e.Seq)
+		}
+	}
+	if n != 1 {
+		t.Fatalf("delivered %d events before close", n)
+	}
+	unsub()
+}
+
+func TestUnsubscribeIsIdempotent(t *testing.T) {
+	b := NewWithBuffer(1)
+	_, unsub := b.Subscribe("c1")
+	unsub()
+	unsub()
+}

internal/config/config.go

@@ -22,6 +22,7 @@ type Config struct {
 	ArtifactsDir string `json:"artifacts"`
 	AuthMode     string `json:"auth"`
 	Mock         bool   `json:"mock"`
+	Instance     string `json:"instance"`
 }
 
 var defaults = Config{
@@ -40,6 +41,7 @@ var envKey = map[string]string{
 	"artifacts": "AGENT_SYSTEM_ARTIFACTS",
 	"auth":      "AGENT_SYSTEM_AUTH",
 	"mock":      "AGENT_SYSTEM_MOCK",
+	"instance":  "AGENT_SYSTEM_INSTANCE",
 }
 
 // Source 配置来源层级（数值越大优先级越高）。
@@ -152,6 +154,8 @@ func loadFile(path string) (Config, map[string]bool, error) {
 			dst = &c.AuthMode
 		case "mock":
 			dst = &c.Mock
+		case "instance":
+			dst = &c.Instance
 		default:
 			return Config{}, nil, fmt.Errorf("config file %s: unknown key %q", path, k)
 		}
@@ -177,6 +181,8 @@ func stringField(c Config, f string) string {
 		return c.ArtifactsDir
 	case "auth":
 		return c.AuthMode
+	case "instance":
+		return c.Instance
 	}
 	return ""
 }
@@ -195,6 +201,8 @@ func setString(c *Config, f, v string) {
 		c.ArtifactsDir = v
 	case "auth":
 		c.AuthMode = v
+	case "instance":
+		c.Instance = v
 	}
 }
 
@@ -216,7 +224,7 @@ func (r Resolved) Explain(w io.Writer, file string) {
 		fmt.Fprintln(w, "config file: (none)")
 	}
 	fmt.Fprintln(w, "field      value                                                    source")
-	for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock"} {
+	for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock", "instance"} {
 		v := stringField(r.Config, f)
 		if f == "mock" {
 			v = fmt.Sprintf("%t", r.Mock)

internal/runmgr/recovery_test.go

@@ -0,0 +1,220 @@
+package runmgr
+
+import (
+	"database/sql"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"agent-system-backend/internal/bus"
+	"agent-system-backend/internal/profiles"
+	"agent-system-backend/internal/settings"
+	"agent-system-backend/internal/store"
+	"agent-system-backend/internal/tools"
+	core "agent-system-core"
+)
+
+func openRecoveryStore(t *testing.T) *store.Store {
+	t.Helper()
+	dsn := os.Getenv("AGENT_PG_TEST_DSN")
+	if dsn == "" {
+		st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "test.db"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { st.Close() })
+		return st
+	}
+	schema := "t_" + strings.ToLower(store.NewID("s")) + "_" + strconv.Itoa(os.Getpid())
+	admin, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
+		admin.Close()
+		t.Fatalf("create schema: %v", err)
+	}
+	t.Cleanup(func() {
+		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
+		admin.Close()
+	})
+	st, err := store.OpenPostgres(dsn+"&search_path="+schema, filepath.Join(t.TempDir(), "artifacts"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { st.Close() })
+	return st
+}
+
+func newRecoveryManager(t *testing.T, st *store.Store) *Manager {
+	t.Helper()
+	reg := tools.NewRegistry()
+	mgr := New(st, bus.New(), reg, profiles.New(st), settings.NewService(st), nil, nil, nil, nil)
+	t.Cleanup(mgr.Shutdown)
+	return mgr
+}
+
+func recoveryConversation(t *testing.T, st *store.Store) core.Conversation {
+	t.Helper()
+	p, err := st.CreateProject(store.OwnerDefault, "p")
+	if err != nil {
+		t.Fatal(err)
+	}
+	c, err := st.CreateConversation(store.OwnerDefault, p.ID, "c", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return c
+}
+
+func recoveryRun(t *testing.T, st *store.Store, conv core.Conversation, status core.RunStatus) core.Run {
+	t.Helper()
+	r := core.Run{
+		ConversationID: conv.ID, ProjectID: conv.ProjectID, Owner: conv.Owner,
+		Kind: core.RunConversation, Status: status, Phase: core.PhaseThinking,
+	}
+	if err := st.CreateRun(&r); err != nil {
+		t.Fatal(err)
+	}
+	return r
+}
+
+func TestReconcileInterruptedSealsOpenRuns(t *testing.T) {
+	st := openRecoveryStore(t)
+	st.SetInstanceID("this")
+	mgr := newRecoveryManager(t, st)
+	mgr.SetInstanceID("this")
+	conv := recoveryConversation(t, st)
+	pending := recoveryRun(t, st, conv, core.RunPending)
+	running := recoveryRun(t, st, conv, core.RunRunning)
+	waiting := recoveryRun(t, st, conv, core.RunWaitingHITL)
+	done := recoveryRun(t, st, conv, core.RunCompleted)
+	card := &core.HITLCard{ID: "card_wait", RunID: waiting.ID, ConversationID: conv.ID, Question: "继续？", Kind: core.HITLKindConfirm}
+	if err := st.SaveHITLCard(conv.Owner, card); err != nil {
+		t.Fatal(err)
+	}
+	if err := mgr.ReconcileInterrupted(); err != nil {
+		t.Fatal(err)
+	}
+	for _, id := range []string{pending.ID, running.ID, waiting.ID} {
+		got, err := st.GetRunByID(id)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Status != core.RunInterrupted || got.EndedAt == nil || got.Error == nil || got.Error.Code != "run.interrupted" {
+			t.Fatalf("run %s: status=%s err=%v ended=%v", id, got.Status, got.Error, got.EndedAt)
+		}
+	}
+	kept, err := st.GetRunByID(done.ID)
+	if err != nil || kept.Status != core.RunCompleted {
+		t.Fatalf("terminal run changed: %+v %v", kept, err)
+	}
+	resolved, err := st.GetHITLCard(conv.Owner, "card_wait")
+	if err != nil || !resolved.Resolved {
+		t.Fatalf("card: %+v %v", resolved, err)
+	}
+	evs, err := st.EventsAfter(conv.ID, 0, 20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	failed := 0
+	for _, e := range evs {
+		if e.Type != core.EvRunFailed {
+			continue
+		}
+		var p core.RunFailedPayload
+		if err := json.Unmarshal(e.Payload, &p); err != nil {
+			t.Fatal(err)
+		}
+		if p.Code != "run.interrupted" {
+			t.Fatalf("payload %+v", p)
+		}
+		failed++
+	}
+	if failed != 3 {
+		t.Fatalf("run.failed events: %d", failed)
+	}
+	if err := mgr.ReconcileInterrupted(); err != nil {
+		t.Fatal(err)
+	}
+	again, err := st.EventsAfter(conv.ID, 0, 20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(again) != len(evs) {
+		t.Fatalf("second reconcile emitted more events: %d -> %d", len(evs), len(again))
+	}
+}
+
+func TestReconcileLeavesOtherInstanceOnPostgres(t *testing.T) {
+	if os.Getenv("AGENT_PG_TEST_DSN") == "" {
+		t.Skip("AGENT_PG_TEST_DSN not set")
+	}
+	st := openRecoveryStore(t)
+	if st.Dialect() != "postgres" {
+		t.Fatal("expected postgres")
+	}
+	mgr := newRecoveryManager(t, st)
+	mgr.SetInstanceID("this")
+	conv := recoveryConversation(t, st)
+	st.SetInstanceID("other")
+	other := recoveryRun(t, st, conv, core.RunRunning)
+	st.SetInstanceID("this")
+	own := recoveryRun(t, st, conv, core.RunRunning)
+	if err := mgr.ReconcileInterrupted(); err != nil {
+		t.Fatal(err)
+	}
+	gotOther, err := st.GetRunByID(other.ID)
+	if err != nil || gotOther.Status != core.RunRunning {
+		t.Fatalf("other instance: %+v %v", gotOther, err)
+	}
+	gotOwn, err := st.GetRunByID(own.ID)
+	if err != nil || gotOwn.Status != core.RunInterrupted {
+		t.Fatalf("own instance: %+v %v", gotOwn, err)
+	}
+}
+
+func TestFinishRunSettleFailureMarksFailed(t *testing.T) {
+	st := openRecoveryStore(t)
+	mgr := newRecoveryManager(t, st)
+	conv := recoveryConversation(t, st)
+	run := recoveryRun(t, st, conv, core.RunRunning)
+	st.SetSettleFault(os.ErrClosed)
+	rc := &runCtx{mgr: mgr, run: run, h: &runHandle{
+		assistant: []core.Part{{Type: core.PartText, Text: "hello"}},
+		started:   time.Now(),
+	}}
+	mgr.finishRun(rc, core.RunCompleted, core.Usage{}, nil)
+	got, err := st.GetRunByID(run.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != core.RunFailed || got.Error == nil || got.Error.Code != "run.settle_failed" {
+		t.Fatalf("status=%s err=%v", got.Status, got.Error)
+	}
+	msgs, err := st.ListMessages(conv.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(msgs) != 0 {
+		t.Fatalf("assistant message must not land when settle fails: %d", len(msgs))
+	}
+	evs, err := st.EventsAfter(conv.ID, 0, 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(evs) != 1 || evs[0].Type != core.EvRunFailed {
+		t.Fatalf("events: %+v", evs)
+	}
+	var p core.RunFailedPayload
+	if err := json.Unmarshal(evs[0].Payload, &p); err != nil {
+		t.Fatal(err)
+	}
+	if p.Code != "run.settle_failed" {
+		t.Fatalf("payload %+v", p)
+	}
+}

internal/runmgr/runmgr.go

@@ -7,9 +7,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"agent-system-backend/internal/bus"
@@ -54,6 +56,9 @@ type Manager struct {
 	convMu     sync.Mutex
 	convActive map[string]bool // conversationID → 有在途 run
 
+	instanceID   string
+	shuttingDown atomic.Bool
+
 	// 后台子代理的完成通知只留在内存里。重启后未投递的通知丢失。
 	noticeMu   sync.Mutex
 	notices    map[string][]bgNotice
@@ -103,6 +108,9 @@ func New(st *store.Store, b *bus.Bus, reg *tools.Registry, p *profiles.Service,
 // Shutdown 停机清空事件缓冲（在 store.Close 之前调用）。
 func (m *Manager) Shutdown() { m.events.stop() }
 
+// SetInstanceID 记录本进程标识。启动对账在 PG 档只收口这个实例的 run。
+func (m *Manager) SetInstanceID(id string) { m.instanceID = id }
+
 // SetEventNotifier 注入落盘后回调（PG 档 pgbus.Notifier；SQLite 档不注入）。
 func (m *Manager) SetEventNotifier(fn func(conversationID string, evs []core.StreamEvent)) {
 	m.events.onPersisted = fn
@@ -799,6 +807,37 @@ func workflowScriptBodies(reg *workflow.Registry, def core.WorkflowDef) map[stri
 	return out
 }
 
+// ClaimConversation 抢占会话槽位。忙则返回 ErrConversationBusy，此时调用方不应写消息。
+func (m *Manager) ClaimConversation(conversationID string) error {
+	return m.claimConversation(conversationID)
+}
+
+// ReleaseConversation 释放已抢到的槽位。启动失败时调用；重复释放无效果。
+func (m *Manager) ReleaseConversation(conversationID string) {
+	m.releaseConversation(conversationID)
+}
+
+// StartClaimedConversationRun 在调用方已经抢到槽位之后启动回合。
+// CreateRun 失败时释放槽位；成功后由 execute 在终态释放。
+func (m *Manager) StartClaimedConversationRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile) (core.Run, error) {
+	return m.startClaimedRun(conv, userMessage, attachments, inputs, nil)
+}
+
+// StartClaimedWorkflowRun 在调用方已经抢到槽位之后启动工作流回合。
+func (m *Manager) StartClaimedWorkflowRun(conv core.Conversation, userMessage core.Message, def core.WorkflowDef, slots map[string]string, missing []workflow.SlotQuestion) (core.Run, error) {
+	copiedSlots := map[string]string{}
+	for k, v := range slots {
+		copiedSlots[k] = v
+	}
+	launch := &workflow.Launch{
+		Def:     workflow.Clone(def),
+		Slots:   copiedSlots,
+		Missing: append([]workflow.SlotQuestion(nil), missing...),
+		Scripts: workflowScriptBodies(m.workflows, def),
+	}
+	return m.startClaimedRun(conv, userMessage, nil, nil, launch)
+}
+
 func (m *Manager) claimConversation(conversationID string) error {
 	m.convMu.Lock()
 	defer m.convMu.Unlock()
@@ -882,27 +921,38 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
 	usage.CompletionTokens = rc.h.completionTokens
 	usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens
 	m.mu.Unlock()
+	if m.shuttingDown.Load() && status == core.RunCancelled {
+		status = core.RunInterrupted
+		errPart = &core.ErrorPart{Code: "run.interrupted", Message: "进程停机，运行已中断"}
+	}
 	now := time.Now().UTC()
 	rc.run.EndedAt = &now
 	rc.run.Status = status
 	rc.run.Phase = core.PhaseDone
 	rc.run.Usage = &usage
 	rc.run.Error = errPart
-	m.st.UpdateRun(&rc.run)
-	if m.OnUsage != nil && usage.TotalTokens > 0 {
-		m.OnUsage(rc.run.Owner, usage.TotalTokens)
-	}
 
-	// 终态权威消息树：落库助手消息（含过程部件标记）。
+	// 终态权威消息树与 run 状态同一次提交。
 	m.mu.Lock()
 	parts := append([]core.Part{}, rc.h.assistant...)
 	m.mu.Unlock()
 	if errPart != nil {
 		parts = append(parts, core.Part{Type: core.PartError, Err: errPart})
 	}
+	var msg *core.Message
 	if len(parts) > 0 {
-		msg := core.Message{ConversationID: rc.run.ConversationID, RunID: rc.run.ID, Role: core.RoleAssistant, Parts: parts}
-		if err := m.st.AppendMessage(&msg); err == nil {
+		msg = &core.Message{ConversationID: rc.run.ConversationID, RunID: rc.run.ID, Role: core.RoleAssistant, Parts: parts}
+	}
+	updated, err := m.st.SettleRun(&rc.run, msg)
+	if err != nil {
+		log.Printf("[runmgr] settle run %s: %v", rc.run.ID, err)
+		m.failSettle(rc, err)
+		return
+	}
+	if m.OnUsage != nil && usage.TotalTokens > 0 {
+		m.OnUsage(rc.run.Owner, usage.TotalTokens)
+	}
+	if msg != nil {
 		// 消息树权威已落库；hitl 摘要部件的卡片状态由前端对账收敛。
 		// submit_card 产物卡回填锚点：卡→消息互定位（D32 方向补全）。
 		m.mu.Lock()
@@ -917,7 +967,6 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
 			}
 		}
 	}
-	}
 
 	// 流式卡兜底（D45）：run 终态时仍有未 commit 的流式卡——按 run 结果置
 	// done/failed，tone 标注「未正常收尾」；SetStatus 恒发布，收件箱可见终态。
@@ -937,6 +986,10 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
 		}
 	}
 
+	// 停机对账已经发出 run.failed 时，这里只补消息，不再发第二次。
+	if !updated {
+		return
+	}
 	switch status {
 	case core.RunCompleted:
 		usagePayload := core.RunCompletedPayload{Usage: usage}
@@ -948,8 +1001,32 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
 		m.mu.Unlock()
 		m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunCompleted, usagePayload)
 	case core.RunFailed:
-		m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: errPart.Code, Message: errPart.Message})
+		code, message := "run.failed", ""
+		if errPart != nil {
+			code, message = errPart.Code, errPart.Message
 		}
+		m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: code, Message: message})
+	case core.RunInterrupted:
+		code, message := "run.interrupted", "进程停机，运行已中断"
+		if errPart != nil && errPart.Code != "" {
+			code, message = errPart.Code, errPart.Message
+		}
+		m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: code, Message: message})
+	}
+}
+
+// failSettle 在终态提交失败后另起一次写入，把 run 标成 failed 并发出 run.failed。
+func (m *Manager) failSettle(rc *runCtx, cause error) {
+	now := time.Now().UTC()
+	rc.run.EndedAt = &now
+	rc.run.Status = core.RunFailed
+	rc.run.Phase = core.PhaseDone
+	rc.run.Error = &core.ErrorPart{Code: "run.settle_failed", Message: cause.Error()}
+	if err := m.st.UpdateRun(&rc.run); err != nil {
+		log.Printf("[runmgr] mark settle failure %s: %v", rc.run.ID, err)
+		return
+	}
+	m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: "run.settle_failed", Message: cause.Error()})
 }
 
 // setRunStatus 更新并广播 phase（状态不单独出事件，phase 事件携带）。
@@ -1328,8 +1405,10 @@ func (m *Manager) startNoticeRun(convID string, items []bgNotice) error {
 	return nil
 }
 
-// InterruptRunning 服务停机前把在途 run 标记为 interrupted（4.2 恢复策略）。
+// InterruptRunning 服务停机前把在途 run 收成 interrupted（4.2 恢复策略）。
+// 与启动对账共用 sealRunInterrupted，终态形状一致。
 func (m *Manager) InterruptRunning() {
+	m.shuttingDown.Store(true)
 	m.mu.Lock()
 	ids := make([]string, 0, len(m.active))
 	for id := range m.active {
@@ -1346,12 +1425,53 @@ func (m *Manager) InterruptRunning() {
 		h.cancel()
 	}
 	for _, id := range ids {
-		r, err := m.st.GetRunByID(id)
-		if err != nil || r.Status == core.RunCompleted || r.Status == core.RunFailed || r.Status == core.RunCancelled {
+		m.sealRunInterrupted(id)
+	}
+}
+
+// ReconcileInterrupted 在开始监听之前调用。
+// SQLite 档扫全部非终态 run。PG 档只扫本实例，以及升级前 instance_id 为空的行。
+func (m *Manager) ReconcileInterrupted() error {
+	all := m.st.Dialect() != "postgres"
+	runs, err := m.st.ListRecoverableRuns(m.instanceID, all)
+	if err != nil {
+		return err
+	}
+	for _, r := range runs {
+		m.sealRunInterrupted(r.ID)
+	}
+	return nil
+}
+
+// sealRunInterrupted 条件更新成功才关 HITL 并发事件。并发调用只有一个赢家。
+func (m *Manager) sealRunInterrupted(runID string) {
+	ok, err := m.st.SealInterrupted(runID)
+	if err != nil || !ok {
+		return
+	}
+	r, err := m.st.GetRunByID(runID)
+	if err != nil {
+		return
+	}
+	m.closePendingHITL(r)
+	code, message := "run.interrupted", "进程重启或停机，运行已中断"
+	if r.Error != nil && r.Error.Code != "" {
+		code, message = r.Error.Code, r.Error.Message
+	}
+	m.emit(r.ConversationID, r.ID, core.EvRunFailed, core.RunFailedPayload{Code: code, Message: message})
+}
+
+func (m *Manager) closePendingHITL(r core.Run) {
+	cards, err := m.st.ActiveHITLCardsByRun(r.ID)
+	if err != nil {
+		return
+	}
+	for _, c := range cards {
+		res := core.HITLResolvedPayload{CardID: c.ID, Kind: "cancelled"}
+		if err := m.st.ResolveHITLCard(r.Owner, string(c.ID), res); err != nil {
 			continue
 		}
-		r.Status = core.RunInterrupted
-		m.st.UpdateRun(&r)
+		m.emit(r.ConversationID, r.ID, core.EvHITLResolved, res)
 	}
 }
 

internal/server/admission_test.go

@@ -0,0 +1,190 @@
+package server
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"agent-system-backend/internal/bus"
+	"agent-system-backend/internal/cmdbus"
+	"agent-system-backend/internal/plugins"
+	"agent-system-backend/internal/profiles"
+	"agent-system-backend/internal/runmgr"
+	"agent-system-backend/internal/runtime"
+	"agent-system-backend/internal/settings"
+	"agent-system-backend/internal/store"
+	"agent-system-backend/internal/tools"
+	core "agent-system-core"
+)
+
+func TestBusySendWritesNothing(t *testing.T) {
+	dir := t.TempDir()
+	st, err := store.OpenSQLite(filepath.Join(dir, "test.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { st.Close() })
+	reg := tools.NewRegistry()
+	reg.RegisterBuiltin()
+	setSvc := settings.NewService(st)
+	profSvc := profiles.New(st)
+	profSvc.EnsureDefault()
+	release := make(chan struct{})
+	blocking := blockingRunner{block: make(chan struct{}, 1), release: release}
+	mgr := runmgr.New(st, bus.New(), reg, profSvc, setSvc, blocking, blocking, &runtime.MockRunner{}, &runtime.MockSubagent{})
+	t.Cleanup(mgr.Shutdown)
+	ph := plugins.New(st, reg, setSvc, filepath.Join(dir, "plugins"))
+	srv := New(st, mgr, reg, profSvc, setSvc, ph)
+	srv.Bus = cmdbus.New(st, nil, nil)
+	srv.WireBuiltinCommands()
+	h := srv.Handler()
+	_, convID := setupConversation(t, h)
+
+	code, _ := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "先占住"})
+	if code != 200 {
+		t.Fatalf("first send: %d", code)
+	}
+	before, err := st.ListMessages(convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	code, _ = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "忙的时候这条不该落库"})
+	if code != http.StatusConflict {
+		t.Fatalf("busy status %d", code)
+	}
+	after, err := st.ListMessages(convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(after) != len(before) {
+		t.Fatalf("busy send wrote a message: %d -> %d", len(before), len(after))
+	}
+	helpCode, helpOut := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "/help"})
+	if helpCode != 200 || helpOut["command"] == nil {
+		t.Fatalf("slash command while busy: %d %v", helpCode, helpOut)
+	}
+	close(release)
+}
+
+func TestRequestIDReturnsExistingMessage(t *testing.T) {
+	srv, st := newTestServer(t)
+	h := srv.Handler()
+	_, convID := setupConversation(t, h)
+	body := map[string]any{"text": "同一请求", "requestId": "req-1"}
+	code, first := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", body)
+	if code != 200 {
+		t.Fatalf("first %d %v", code, first)
+	}
+	code, second := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", body)
+	if code != 200 {
+		t.Fatalf("second %d %v", code, second)
+	}
+	m1 := first["message"].(map[string]any)["id"]
+	m2 := second["message"].(map[string]any)["id"]
+	if m1 != m2 {
+		t.Fatalf("message ids %v %v", m1, m2)
+	}
+	msgs, err := st.ListMessages(convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	user := 0
+	for _, m := range msgs {
+		if m.Role == core.RoleUser {
+			user++
+		}
+	}
+	if user != 1 {
+		t.Fatalf("user messages: %d", user)
+	}
+	runs, err := st.ListRunsByConversation(convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(runs) != 1 {
+		t.Fatalf("runs: %d", len(runs))
+	}
+}
+
+func TestSSEReplaysBacklogBeyondOnePage(t *testing.T) {
+	srv, st := newTestServer(t)
+	h := srv.Handler()
+	_, convID := setupConversation(t, h)
+	const total = 2500
+	batch := make([]core.StreamEvent, 0, 500)
+	for i := 0; i < total; i++ {
+		batch = append(batch, core.StreamEvent{
+			ConversationID: convID, RunID: "r", TS: time.Now().UTC().Format(time.RFC3339Nano),
+			V: core.ProtocolVersion, Type: "note",
+		})
+		if len(batch) == 500 {
+			if _, err := st.AppendEvents(convID, batch); err != nil {
+				t.Fatal(err)
+			}
+			batch = batch[:0]
+		}
+	}
+	srv.Subscribe = func(string) (<-chan core.StreamEvent, func()) {
+		ch := make(chan core.StreamEvent)
+		close(ch)
+		return ch, func() {}
+	}
+	req := httptest.NewRequest(http.MethodGet, "/api/events?conversationId="+convID, nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	ids := sseIDs(rec.Body.String())
+	if len(ids) != total {
+		t.Fatalf("replayed %d, want %d", len(ids), total)
+	}
+	for i, id := range ids {
+		if id != i+1 {
+			t.Fatalf("seq[%d]=%d", i, id)
+		}
+	}
+}
+
+func TestSSEReconnectFillsMissingSeq(t *testing.T) {
+	srv, st := newTestServer(t)
+	h := srv.Handler()
+	_, convID := setupConversation(t, h)
+	batch := make([]core.StreamEvent, 5)
+	for i := range batch {
+		batch[i] = core.StreamEvent{
+			ConversationID: convID, RunID: "r", TS: time.Now().UTC().Format(time.RFC3339Nano),
+			V: core.ProtocolVersion, Type: "note",
+		}
+	}
+	if _, err := st.AppendEvents(convID, batch); err != nil {
+		t.Fatal(err)
+	}
+	srv.Subscribe = func(string) (<-chan core.StreamEvent, func()) {
+		ch := make(chan core.StreamEvent)
+		close(ch)
+		return ch, func() {}
+	}
+	req := httptest.NewRequest(http.MethodGet, "/api/events?conversationId="+convID+"&since=2", nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	ids := sseIDs(rec.Body.String())
+	if len(ids) != 3 || ids[0] != 3 || ids[2] != 5 {
+		t.Fatalf("reconnect seqs %v", ids)
+	}
+}
+
+func sseIDs(body string) []int {
+	var ids []int
+	for _, line := range strings.Split(body, "\n") {
+		if !strings.HasPrefix(line, "id: ") {
+			continue
+		}
+		n := 0
+		for _, c := range line[4:] {
+			n = n*10 + int(c-'0')
+		}
+		ids = append(ids, n)
+	}
+	return ids
+}

internal/server/server.go

@@ -310,8 +310,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 	}
 
 	lastSent := since
-	replay, err := s.st.EventsAfter(conversationID, since, 1000)
-	if err == nil {
+	for {
+		replay, err := s.st.EventsAfter(conversationID, lastSent, 1000)
+		if err != nil || len(replay) == 0 {
+			break
+		}
 		for _, e := range replay {
 			if e.Seq <= lastSent {
 				continue
@@ -322,8 +325,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 			}
 			lastSent = e.Seq
 		}
+		if len(replay) < 1000 {
+			break
 		}
-	// 心跳 + 在线转发。
+	}
+	// 心跳 + 在线转发。通道关闭表示缓冲溢出，结束响应让 EventSource 带 Last-Event-ID 重连。
 	heartbeat := time.NewTicker(15 * time.Second)
 	defer heartbeat.Stop()
 	for {
@@ -334,7 +340,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 		case <-heartbeat.C:
 			fmt.Fprint(w, ": ping\n\n")
 			flusher.Flush()
-		case e := <-live:
+		case e, ok := <-live:
+			if !ok {
+				stop()
+				return
+			}
 			if e.ConversationID != conversationID || e.Seq <= lastSent {
 				continue
 			}
@@ -523,6 +533,7 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
 	var req struct {
 		Text          string   `json:"text"`
 		AttachmentIDs []string `json:"attachmentIds"`
+		RequestID     string   `json:"requestId"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
@@ -533,6 +544,17 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "text or attachments required", http.StatusBadRequest)
 		return
 	}
+	if req.RequestID != "" {
+		existing, err := s.st.MessageByRequest(convID, req.RequestID)
+		if err == nil {
+			s.writeExistingMessage(w, convID, existing)
+			return
+		}
+		if !errors.Is(err, store.ErrNotFound) {
+			writeStoreErr(w, err)
+			return
+		}
+	}
 
 	// 技能命令：提示词库命令触发命中时保持原文、跳过命令路由，走正常 run——
 	// system 组装时命令触发照常命中（命令是显式意图，不受附件门槛约束）；
@@ -570,15 +592,46 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
 		parsedDocs = append(parsedDocs, doc)
 		parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}})
 	}
-	msg := core.Message{ConversationID: convID, Role: core.RoleUser, Parts: parts}
-	if err := s.st.AppendMessage(&msg); err != nil {
+
+	// 先路由再落库。回合和工作流要先抢槽位，忙则 409 且不写任何数据。
+	// 意图命中的命令不占槽位，会话忙时也能用。
+	route := "turn"
+	var recognized intentgate.MatchResult
+	if s.Gate != nil && !strings.HasPrefix(text, "/") {
+		recognized = s.recognize(r.Context(), text)
+		if recognized.Branch == "matched" && s.Bus != nil && s.Bus.HasCommand(recognized.ID) {
+			route = "command"
+		} else if recognized.Branch == "matched" && s.Workflows != nil {
+			if _, ok := s.Workflows.ByIntent(recognized.ID); ok {
+				route = "workflow"
+			}
+		}
+	}
+	claimed := false
+	if route == "turn" || route == "workflow" {
+		if err := s.mgr.ClaimConversation(convID); err != nil {
 			writeStoreErr(w, err)
 			return
 		}
-	// 先命令总线，再绑定工作流，最后才是普通回合。
-	if s.Gate != nil && !strings.HasPrefix(text, "/") {
-		recognized := s.recognize(r.Context(), text)
-		if recognized.Branch == "matched" && s.Bus != nil && s.Bus.HasCommand(recognized.ID) {
+		claimed = true
+	}
+	msg := core.Message{ConversationID: convID, Role: core.RoleUser, Parts: parts}
+	if err := s.st.AppendMessageWithRequest(&msg, req.RequestID); err != nil {
+		var dup *store.DuplicateRequestError
+		if errors.As(err, &dup) {
+			if claimed {
+				s.mgr.ReleaseConversation(convID)
+			}
+			s.writeExistingMessage(w, convID, dup.Existing)
+			return
+		}
+		if claimed {
+			s.mgr.ReleaseConversation(convID)
+		}
+		writeStoreErr(w, err)
+		return
+	}
+	if route == "command" {
 		resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{
 			ID: recognized.ID, Params: map[string]any{"text": text}, Entrance: "chat", Actor: owner,
 			Owner: owner, ConversationID: convID, AnchorMessageID: msg.ID,
@@ -595,20 +648,42 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
 		writeJSON(w, http.StatusOK, out)
 		return
 	}
-		if recognized.Branch == "matched" && s.Workflows != nil {
-			if def, ok := s.Workflows.ByIntent(recognized.ID); ok {
+	var run core.Run
+	var startErr error
+	if route == "workflow" {
+		def, _ := s.Workflows.ByIntent(recognized.ID)
 		missing := make([]workflow.SlotQuestion, len(recognized.Missing))
 		for i, q := range recognized.Missing {
 			missing[i] = workflow.SlotQuestion{Key: q.Key, Question: q.Question}
 		}
-				run, err := s.mgr.StartWorkflowRun(conv, msg, def, recognized.Slots, missing)
-				writeJSONOr(w, map[string]any{"message": msg, "run": run}, err)
+		run, startErr = s.mgr.StartClaimedWorkflowRun(conv, msg, def, recognized.Slots, missing)
+	} else {
+		run, startErr = s.mgr.StartClaimedConversationRun(conv, msg, parsedDocs, inputs)
+	}
+	if startErr != nil {
+		s.mgr.ReleaseConversation(convID)
+		writeStoreErr(w, startErr)
+		return
+	}
+	_ = s.st.SetMessageRunID(msg.ID, run.ID)
+	msg.RunID = run.ID
+	writeJSON(w, http.StatusOK, map[string]any{"message": msg, "run": run})
+}
+
+// writeExistingMessage 把同一 requestId 的已有消息和仍在进行的 run 原样返回。
+func (s *Server) writeExistingMessage(w http.ResponseWriter, convID string, msg core.Message) {
+	out := map[string]any{"message": msg}
+	if msg.RunID != "" {
+		if run, err := s.st.GetRunByID(msg.RunID); err == nil {
+			out["run"] = run
+			writeJSON(w, http.StatusOK, out)
 			return
 		}
 	}
+	if run, err := s.st.LatestOpenRun(convID); err == nil {
+		out["run"] = run
 	}
-	run, err := s.mgr.StartConversationRun(conv, msg, parsedDocs, inputs)
-	writeJSONOr(w, map[string]any{"message": msg, "run": run}, err)
+	writeJSON(w, http.StatusOK, out)
 }
 
 // handleCommand `/` 入口 → 命令总线（D27 四入口同源；slash 语义解析后统一分发+审计）。

internal/server/workflow_route_test.go

@@ -106,6 +106,48 @@ func TestIntentStartsWorkflowRun(t *testing.T) {
 	}
 }
 
+func TestWorkflowRouteClaimsBeforePersist(t *testing.T) {
+	srv, st := newTestServer(t)
+	if err := st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": false}}); err != nil {
+		t.Fatal(err)
+	}
+	gate := intentgate.New()
+	if err := gate.Rebuild(context.Background(), []json.RawMessage{briefIntent()}); err != nil {
+		t.Fatal(err)
+	}
+	srv.Gate = gate
+	reg := workflow.NewRegistry()
+	reg.Set([]core.WorkflowDef{{
+		ID: "demo.brief", Title: "简报", IntentID: "demo.brief",
+		Steps: []core.WorkflowStep{{ID: "s", Kind: "hitl"}},
+	}}, nil)
+	srv.Workflows = reg
+	_, convID := setupConversation(t, srv.Handler())
+	conv, err := st.GetConversation(store.OwnerDefault, convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := srv.mgr.ClaimConversation(convID); err != nil {
+		t.Fatal(err)
+	}
+	before, err := st.ListMessages(convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	code, out := doJSON(t, srv.Handler(), "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "写一份简报"})
+	if code != 409 {
+		t.Fatalf("status %d body %v", code, out)
+	}
+	after, err := st.ListMessages(convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(after) != len(before) {
+		t.Fatalf("workflow route wrote a message while busy")
+	}
+	srv.mgr.ReleaseConversation(conv.ID)
+}
+
 func briefIntent() json.RawMessage {
 	return json.RawMessage(`{"id":"demo.brief","domain":"demo","version":1,"matcher":{"includes_any":["简报"]},"slots":[{"key":"topic","required":true,"pattern":"主题[:：]\\s*(\\S+)","question":"主题是什么？"}]}`)
 }

internal/store/messages.go

@@ -28,14 +28,20 @@ func isUniqueViolation(err error) bool {
 // AppendMessage 以单语句原子分配会话内序号（INSERT ... SELECT MAX+1）；
 // 并发同会话追加由 uq_messages_conv_seq 唯一约束兜底，冲突有界重试。
 func (s *Store) AppendMessage(m *core.Message) error {
+	return s.AppendMessageWithRequest(m, "")
+}
+
+// AppendMessageWithRequest 同 AppendMessage，并记录可选的 request_id。
+// 同一会话里非空 request_id 已存在时，把 *m 换成已有行并返回 DuplicateRequestError。
+func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) error {
 	if m.ID == "" {
 		m.ID = newID("msg")
 	}
 	m.CreatedAt = m.CreatedAt.UTC()
-	insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at)
-		SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,? FROM messages WHERE conversation_id=?`
+	insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id)
+		SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?`
 	for attempt := 0; attempt < 8; attempt++ {
-		_, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), m.ConversationID)
+		_, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID)
 		if err == nil {
 			s.TouchConversation(m.ConversationID)
 			return nil
@@ -43,6 +49,13 @@ func (s *Store) AppendMessage(m *core.Message) error {
 		if !isUniqueViolation(err) {
 			return err
 		}
+		if requestID != "" {
+			existing, lookErr := s.MessageByRequest(string(m.ConversationID), requestID)
+			if lookErr == nil {
+				*m = existing
+				return &DuplicateRequestError{Existing: existing}
+			}
+		}
 		// 并发同会话追加冲突（PG 多连接竞态）：短抖动退避后重算序号。
 		time.Sleep(time.Duration(attempt) * time.Millisecond)
 	}
@@ -106,8 +119,8 @@ func (s *Store) CreateRun(r *core.Run) error {
 		r.ID = newID("run")
 	}
 	r.CreatedAt = r.CreatedAt.UTC()
-	_, err := s.db.Exec(s.q(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)`),
-		r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO())
+	_, err := s.db.Exec(s.q(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at,instance_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`),
+		r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO(), s.instanceID)
 	return err
 }
 

internal/store/migrate.go

@@ -369,6 +369,27 @@ func init() {
 					`DROP TABLE IF EXISTS plugin_grants`,
 				},
 			}},
+		migration{version: 14, name: "run_instance_and_request_id",
+			// 启动对账按实例收口在途 run；发消息按 request_id 去重。
+			// 空 request_id 不参与唯一约束，旧消息与命令路径仍可重复追加。
+			up: map[string][]string{"sqlite": {
+				`ALTER TABLE runs ADD COLUMN instance_id TEXT NOT NULL DEFAULT ''`,
+				`ALTER TABLE messages ADD COLUMN request_id TEXT NOT NULL DEFAULT ''`,
+				`CREATE UNIQUE INDEX IF NOT EXISTS uq_messages_conv_request ON messages(conversation_id, request_id) WHERE request_id <> ''`,
+			}, "postgres": {
+				`ALTER TABLE runs ADD COLUMN instance_id TEXT NOT NULL DEFAULT ''`,
+				`ALTER TABLE messages ADD COLUMN request_id TEXT NOT NULL DEFAULT ''`,
+				`CREATE UNIQUE INDEX IF NOT EXISTS uq_messages_conv_request ON messages(conversation_id, request_id) WHERE request_id <> ''`,
+			}},
+			down: map[string][]string{"sqlite": {
+				`DROP INDEX IF EXISTS uq_messages_conv_request`,
+				`ALTER TABLE messages DROP COLUMN request_id`,
+				`ALTER TABLE runs DROP COLUMN instance_id`,
+			}, "postgres": {
+				`DROP INDEX IF EXISTS uq_messages_conv_request`,
+				`ALTER TABLE messages DROP COLUMN request_id`,
+				`ALTER TABLE runs DROP COLUMN instance_id`,
+			}}},
 	)
 }
 

internal/store/recovery.go

@@ -0,0 +1,188 @@
+package store
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+
+	core "agent-system-core"
+)
+
+// DuplicateRequestError 带上已落库的那条消息。
+type DuplicateRequestError struct {
+	Existing core.Message
+}
+
+func (e *DuplicateRequestError) Error() string { return ErrDuplicateRequest.Error() }
+
+func (e *DuplicateRequestError) Unwrap() error { return ErrDuplicateRequest }
+
+// MessageByRequest 按会话和 request_id 查找已落库消息。
+func (s *Store) MessageByRequest(conversationID, requestID string) (core.Message, error) {
+	if requestID == "" {
+		return core.Message{}, ErrNotFound
+	}
+	var m core.Message
+	var parts, ca string
+	err := s.db.QueryRow(s.q(`SELECT id,conversation_id,run_id,role,parts,created_at FROM messages WHERE conversation_id=? AND request_id=?`),
+		conversationID, requestID).Scan(&m.ID, &m.ConversationID, &m.RunID, &m.Role, &parts, &ca)
+	if errors.Is(err, sql.ErrNoRows) {
+		return core.Message{}, ErrNotFound
+	}
+	if err != nil {
+		return core.Message{}, err
+	}
+	m.Role = core.Role(m.Role)
+	m.Parts, _ = unmarshalJSON[[]core.Part](parts)
+	m.CreatedAt, _ = parseTime(ca)
+	return m, nil
+}
+
+// SetMessageRunID 把用户消息挂到随后启动的 run 上，供同一 request_id 重试时带回。
+func (s *Store) SetMessageRunID(messageID, runID string) error {
+	_, err := s.db.Exec(s.q(`UPDATE messages SET run_id=? WHERE id=?`), runID, messageID)
+	return err
+}
+
+// LatestOpenRun 返回会话里最新的非终态、非子代理 run。
+// 幂等重试在用户消息还没写上 run_id 时，用它避免绑到更早的已完成 run。
+func (s *Store) LatestOpenRun(conversationID string) (core.Run, error) {
+	r, err := scanRun(func(dst ...any) error {
+		return s.db.QueryRow(s.q(`SELECT `+runColumns+` FROM runs WHERE conversation_id=? AND kind<>? AND status IN ('pending','running','waiting_hitl') ORDER BY created_at DESC LIMIT 1`),
+			conversationID, string(core.RunSubagent)).Scan(dst...)
+	})
+	if errors.Is(err, sql.ErrNoRows) {
+		return r, ErrNotFound
+	}
+	return r, err
+}
+
+// ListRecoverableRuns 列出仍停在非终态的 run。
+// all 为真时扫全部（SQLite 单进程）。否则只扫本实例，以及升级前 instance_id 为空的行。
+func (s *Store) ListRecoverableRuns(instanceID string, all bool) ([]core.Run, error) {
+	q := `SELECT ` + runColumns + ` FROM runs WHERE status IN ('pending','running','waiting_hitl')`
+	args := []any{}
+	if !all {
+		q += ` AND (instance_id=? OR instance_id='')`
+		args = append(args, instanceID)
+	}
+	q += ` ORDER BY created_at`
+	rows, err := s.db.Query(s.q(q), args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []core.Run{}
+	for rows.Next() {
+		r, err := scanRun(rows.Scan)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+// SealInterrupted 把仍在进行的 run 收成 interrupted。
+// 条件更新保证并发对账时只有一个赢家；已经是终态的行不受影响。
+func (s *Store) SealInterrupted(id string) (bool, error) {
+	errPart := marshalJSON(&core.ErrorPart{Code: "run.interrupted", Message: "进程重启或停机，运行已中断"})
+	res, err := s.db.Exec(s.q(`UPDATE runs SET status=?, phase=?, error=?, ended_at=? WHERE id=? AND status IN ('pending','running','waiting_hitl')`),
+		string(core.RunInterrupted), string(core.PhaseDone), errPart, nowISO(), id)
+	if err != nil {
+		return false, err
+	}
+	n, err := res.RowsAffected()
+	return n == 1, err
+}
+
+// ActiveHITLCardsByRun 返回某个 run 上尚未关闭的 HITL 卡片。
+func (s *Store) ActiveHITLCardsByRun(runID string) ([]core.HITLCard, error) {
+	rows, err := s.db.Query(s.q(`SELECT payload FROM hitl_cards WHERE run_id=? AND resolved=0 ORDER BY created_at`), runID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []core.HITLCard{}
+	for rows.Next() {
+		var payload string
+		if err := rows.Scan(&payload); err != nil {
+			return nil, err
+		}
+		c, err := unmarshalJSON[core.HITLCard](payload)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}
+
+// SettleRun 在一个事务里更新 run，并可选地追加助手消息。
+// updated 为假表示条件更新没有改到行（已被收成终态）；消息仍会写入。
+// 序号冲突时整段事务重试。其它失败回滚，run 保持原状态。
+func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err error) {
+	s.settleMu.Lock()
+	fault := s.settleFault
+	s.settleMu.Unlock()
+	if fault != nil {
+		return false, fault
+	}
+	if msg != nil && msg.ID == "" {
+		msg.ID = newID("msg")
+	}
+	var last error
+	for attempt := 0; attempt < 8; attempt++ {
+		updated, last = s.settleRunOnce(r, msg)
+		if last == nil {
+			return updated, nil
+		}
+		if !isUniqueViolation(last) {
+			return false, last
+		}
+		time.Sleep(time.Duration(attempt) * time.Millisecond)
+	}
+	if last == nil {
+		last = fmt.Errorf("store: settle retries exhausted (run %s)", r.ID)
+	}
+	return false, last
+}
+
+func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return false, err
+	}
+	defer tx.Rollback()
+	if err := s.dialect.lockConversation(tx, string(r.ConversationID)); err != nil {
+		return false, err
+	}
+	var ended any
+	if r.EndedAt != nil {
+		ended = nowISO()
+	}
+	res, err := tx.Exec(s.q(`UPDATE runs SET status=?, phase=?, usage=?, error=?, ended_at=? WHERE id=? AND status IN ('pending','running','waiting_hitl')`),
+		string(r.Status), string(r.Phase), marshalJSON(r.Usage), marshalJSON(r.Error), ended, r.ID)
+	if err != nil {
+		return false, err
+	}
+	n, err := res.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	if msg != nil {
+		insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id)
+			SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?`
+		if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), nowISO(), "", msg.ConversationID); err != nil {
+			return false, err
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		return false, err
+	}
+	if msg != nil {
+		s.TouchConversation(string(msg.ConversationID))
+	}
+	return n == 1, nil
+}

internal/store/recovery_test.go

@@ -0,0 +1,134 @@
+package store
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	core "agent-system-core"
+)
+
+func seedConversation(t *testing.T, st *Store) core.Conversation {
+	t.Helper()
+	p, err := st.CreateProject(OwnerDefault, "p")
+	if err != nil {
+		t.Fatal(err)
+	}
+	c, err := st.CreateConversation(OwnerDefault, p.ID, "c", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return c
+}
+
+func seedRun(t *testing.T, st *Store, conv core.Conversation, status core.RunStatus) core.Run {
+	t.Helper()
+	r := core.Run{
+		ConversationID: conv.ID, ProjectID: conv.ProjectID, Owner: conv.Owner,
+		Kind: core.RunConversation, Status: status, Phase: core.PhaseThinking,
+	}
+	if err := st.CreateRun(&r); err != nil {
+		t.Fatal(err)
+	}
+	return r
+}
+
+func TestMessageRequestIDDedup(t *testing.T) {
+	st := openTestStore(t)
+	conv := seedConversation(t, st)
+	first := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi"}}}
+	if err := st.AppendMessageWithRequest(&first, "req-1"); err != nil {
+		t.Fatal(err)
+	}
+	second := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi again"}}}
+	err := st.AppendMessageWithRequest(&second, "req-1")
+	var dup *DuplicateRequestError
+	if !errors.As(err, &dup) || dup.Existing.ID != first.ID {
+		t.Fatalf("duplicate request: %v %#v", err, dup)
+	}
+	blankA := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "a"}}}
+	blankB := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "b"}}}
+	if err := st.AppendMessageWithRequest(&blankA, ""); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.AppendMessageWithRequest(&blankB, ""); err != nil {
+		t.Fatal(err)
+	}
+	msgs, err := st.ListMessages(conv.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(msgs) != 3 {
+		t.Fatalf("messages: %d", len(msgs))
+	}
+}
+
+func TestCreateRunRecordsInstance(t *testing.T) {
+	st := openTestStore(t)
+	st.SetInstanceID("host:8787")
+	conv := seedConversation(t, st)
+	run := seedRun(t, st, conv, core.RunPending)
+	var got string
+	if err := st.db.QueryRow(st.q(`SELECT instance_id FROM runs WHERE id=?`), run.ID).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	if got != "host:8787" {
+		t.Fatalf("instance_id %q", got)
+	}
+}
+
+func TestRecoverableRunsRespectInstance(t *testing.T) {
+	st := openTestStore(t)
+	conv := seedConversation(t, st)
+	st.SetInstanceID("inst-a")
+	own := seedRun(t, st, conv, core.RunPending)
+	st.SetInstanceID("inst-b")
+	other := seedRun(t, st, conv, core.RunRunning)
+	st.SetInstanceID("")
+	legacy := seedRun(t, st, conv, core.RunWaitingHITL)
+	done := seedRun(t, st, conv, core.RunCompleted)
+
+	all := st.Dialect() != "postgres"
+	got, err := st.ListRecoverableRuns("inst-a", all)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ids := map[string]bool{}
+	for _, r := range got {
+		ids[r.ID] = true
+	}
+	if !ids[own.ID] || !ids[legacy.ID] || ids[done.ID] {
+		t.Fatalf("recoverable ids: %v", ids)
+	}
+	if st.Dialect() == "postgres" && ids[other.ID] {
+		t.Fatal("postgres reconcile must leave another instance's run alone")
+	}
+	if st.Dialect() != "postgres" && !ids[other.ID] {
+		t.Fatal("sqlite reconcile scans every open run")
+	}
+}
+
+func TestSettleRunRollsBackWhenMessageInsertFails(t *testing.T) {
+	st := openTestStore(t)
+	conv := seedConversation(t, st)
+	run := seedRun(t, st, conv, core.RunRunning)
+	existing := core.Message{ID: "msg_fixed", ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "user"}}}
+	if err := st.AppendMessage(&existing); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now().UTC()
+	run.Status = core.RunCompleted
+	run.Phase = core.PhaseDone
+	run.EndedAt = &now
+	dup := core.Message{ID: "msg_fixed", ConversationID: conv.ID, RunID: run.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "assistant"}}}
+	if _, err := st.SettleRun(&run, &dup); err == nil {
+		t.Fatal("duplicate message id must fail the settle")
+	}
+	got, err := st.GetRunByID(run.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != core.RunRunning {
+		t.Fatalf("rolled-back status: %s", got.Status)
+	}
+}

internal/store/store.go

@@ -16,6 +16,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"time"
 
 	"agent-system-backend/internal/blobstore"
@@ -27,6 +28,10 @@ import (
 var ErrNotFound = errors.New("store: not found")
 var ErrConflict = errors.New("store: conflict")
 
+// ErrDuplicateRequest 表示同一会话的 request_id 已经落库。
+// 调用方应返回已有消息，不要映射成 409：客户端把 409 当成会话忙并重新入队。
+var ErrDuplicateRequest = errors.New("store: duplicate request id")
+
 // OwnerDefault 是单租户默认 owner（嵌入式档 / 未启用认证时的隐式身份）。
 const OwnerDefault = "local"
 
@@ -35,6 +40,10 @@ type Store struct {
 	dialect    dialect
 	dataDir    string          // 工件对象目录（备份路径等诊断用）
 	blobs      blobstore.Store // 工件对象存储（本地档；多实例切共享卷/对象存储）
+	instanceID string          // 本进程标识；CreateRun 写入 runs.instance_id
+
+	settleMu    sync.Mutex
+	settleFault error // 非空时 SettleRun 在写入前失败，用于终态提交失败测试
 }
 
 // OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。
@@ -101,6 +110,17 @@ func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Stor
 
 func (s *Store) Close() error { return s.db.Close() }
 
+// SetInstanceID 记录本进程标识。之后 CreateRun 把它写入 runs.instance_id。
+// 只在开始接请求前调用一次。
+func (s *Store) SetInstanceID(id string) { s.instanceID = id }
+
+// SetSettleFault 让下一次 SettleRun 在写入前失败。nil 表示恢复正常。
+func (s *Store) SetSettleFault(err error) {
+	s.settleMu.Lock()
+	s.settleFault = err
+	s.settleMu.Unlock()
+}
+
 // Dialect 返回当前存储档名（sqlite | postgres；诊断/健康检查用）。
 func (s *Store) Dialect() string { return s.dialect.name() }
 

internal/store/store_test.go

@@ -56,8 +56,8 @@ func TestMigrationUpgradesExistingDatabase(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(got) != 13 {
-		t.Fatalf("expected versions [1..13], got %v", got)
+	if len(got) != 14 {
+		t.Fatalf("expected versions [1..14], got %v", got)
 	}
 	// 老库形态（v2 前）：events 全局自增、settings 单列主键。
 	if err := st.MigrateDown(1); err != nil {