cmd/storage-migrate/main_test.go

@@ -171,7 +171,11 @@ func TestCopyOrderCoversAllFixtureTables(t *testing.T) {
 		known[t] = true
 	}
 	known["schema_migrations"] = true // 迁移器管理，不随迁
-	known["admission_leases"] = true  // 运行态，明确排除（见 copyOrder 注释）
+	// 运行态表不随迁：执行许可是进程瞬态；执行租约与跨实例命令属当前
+	// 实例的调度现场（接管语义在 U5 内由 lease/fencing 处理，不复制历史）。
+	for _, rt := range []string{"admission_leases", "conversation_executions", "run_commands"} {
+		known[rt] = true
+	}
 	// FTS5 虚表影子表：SQLite 内部结构，PG 侧由 v4 生成列/索引从 knowledge
 	// 行自动重建，不随迁。
 	for _, shadow := range []string{"knowledge_fts", "knowledge_fts_data", "knowledge_fts_idx", "knowledge_fts_docsize", "knowledge_fts_config"} {

internal/app/app.go

@@ -94,6 +94,23 @@ type Options struct {
 	PGPool store.PoolOptions
 }
 
+// instanceTag 稳定执行身份：tier=team 时强制 host:addr（多实例互异），
+// 本地档退化为 local（单实例语义）。
+func instanceTag(opt Options) string {
+	if opt.Tier == "team" || opt.StorageDSN != "" {
+		return hostTag() + ":" + opt.Addr
+	}
+	return "local"
+}
+
+func hostTag() string {
+	h, err := os.Hostname()
+	if err != nil || h == "" {
+		return "host"
+	}
+	return h
+}
+
 // tierName 输出部署档名（诊断统一拼写：空 = auto）。
 func tierName(opt Options) string {
 	if opt.Tier == "" {
@@ -481,6 +498,11 @@ func Run(opt Options) error {
 		}
 	}()
 	mgr.SetCards(cardSvc)
+	// 执行身份与命令消费（协作存储 U5）：PG 档启动跨实例命令轮询；
+	// SQLite 本地档无跨实例命令（StartCommandConsumer 自行短路）。
+	mgr.SetInstanceTag(instanceTag(opt), opt.Addr)
+	stopCommands := mgr.StartCommandConsumer(context.Background())
+	defer stopCommands()
 	// 执行预算（协作存储 U3，R8）：runtime.execBudgetGlobal / runtime.execBudgetPerUser
 	// 全零=不限（本地档默认，行为不变）；开启后执行许可共享账本生效，
 	// 待执行队列上限 runtime.execQueueMax（默认 200，容量合同数值待 U7 实测调整）。

internal/runmgr/commands.go

@@ -0,0 +1,178 @@
+package runmgr
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"time"
+
+	core "agent-system-core"
+)
+
+// 协作存储 U5（KTD5）：持久命令消费。任意 API 实例把取消/HITL 写入
+// run_commands 收件箱；命令由「目标 run 的当前执行实例」消费（路由依据
+// conversation_executions 的当前 owner；本地不在执行的实例跳过，命令保持
+// pending 直到执行实例领取或超时由恢复路径收口）。
+//
+// 本文件是执行侧消费循环；server 侧的投递入口见 server 包 EnqueueRunCommand
+// 的两个薄封装（CancelRunRemote / ResolveHITLRemote）。
+
+// commandPollEvery 消费轮询间隔。
+const commandPollEvery = 300 * time.Millisecond
+
+// HITLCommandPayload HITL 命令载荷（gate 身份由 cardID+gateVersion 承载：
+// 新 owner 消费时重验 gate 版本，过期审批明确拒绝而不静默丢弃——KTD5）。
+type HITLCommandPayload struct {
+	Owner       string          `json:"owner"`
+	CardID      string          `json:"cardId"`
+	GateVersion int64           `json:"gateVersion"`
+	Kind        string          `json:"kind"`
+	Answer      json.RawMessage `json:"answer,omitempty"`
+}
+
+// CancelCommandPayload 取消命令载荷。
+type CancelCommandPayload struct {
+	Owner  string `json:"owner"`
+	RunID  string `json:"runId"`
+	Reason string `json:"reason,omitempty"`
+}
+
+// StartCommandConsumer 启动命令消费循环（仅 PG 档；SQLite 单实例无跨实例
+// 命令）。返回停止函数。消费的命令只处理「本实例当前持有执行权」的会话；
+// 其余跳过（不认领）——避免非执行实例吃掉命令。
+func (m *Manager) StartCommandConsumer(ctx context.Context) (stop func()) {
+	if m.st.Dialect() != "postgres" {
+		return func() {}
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	go func() {
+		t := time.NewTicker(commandPollEvery)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				m.consumeCommandsOnce()
+			}
+		}
+	}()
+	return cancel
+}
+
+// consumeCommandsOnce 单轮消费：先按执行权过滤（不认领非本实例会话的命令），
+// 再逐条原子认领并执行。认领即消费（at-least-once；命令处理幂等：取消对
+// 终态 run 报已终态、HITL 重验 gate）。
+func (m *Manager) consumeCommandsOnce() {
+	pending, err := m.pendingCommandsForOwned()
+	if err != nil || len(pending) == 0 {
+		return
+	}
+	for _, c := range pending {
+		claimed, err := m.claimCommand(c.CommandID)
+		if err != nil || !claimed {
+			continue
+		}
+		switch c.Kind {
+		case "cancel":
+			m.applyCancelCommand(c)
+		case "hitl":
+			m.applyHITLCommand(c)
+		default:
+			log.Printf("[runmgr] command %s: unknown kind %q", c.CommandID, c.Kind)
+		}
+	}
+}
+
+// pendingCommandsForOwned 列出 pending 且其会话执行权在本实例的命令。
+func (m *Manager) pendingCommandsForOwned() ([]ownedCommand, error) {
+	rows, err := m.st.DB().Query(
+		`SELECT c.command_id, c.kind, c.run_id, c.conversation_id, c.epoch, c.payload
+		 FROM run_commands c JOIN conversation_executions e ON e.conversation_id = c.conversation_id
+		 WHERE c.state='pending' AND e.instance_id = $1
+		 ORDER BY c.created_at LIMIT 16`, m.InstanceID())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []ownedCommand
+	for rows.Next() {
+		var c ownedCommand
+		if err := rows.Scan(&c.CommandID, &c.Kind, &c.RunID, &c.ConversationID, &c.Epoch, &c.Payload); err != nil {
+			return nil, err
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}
+
+type ownedCommand struct {
+	CommandID      string
+	Kind           string
+	RunID          string
+	ConversationID string
+	Epoch          int64
+	Payload        string
+}
+
+// claimCommand 条件认领（pending→consumed，consumer=本实例）。
+func (m *Manager) claimCommand(commandID string) (bool, error) {
+	res, err := m.st.DB().Exec(
+		`UPDATE run_commands SET state='consumed', consumed_at=$3, consumer=$1 WHERE command_id=$2 AND state='pending'`,
+		m.InstanceID(), commandID, time.Now().UTC().Format(time.RFC3339Nano))
+	if err != nil {
+		return false, err
+	}
+	n, _ := res.RowsAffected()
+	return n > 0, nil
+}
+
+// applyCancelCommand 本地执行取消：run 在本实例活跃 → 级联取消；
+// 不活跃（如执行实例重启后命令迟到）→ 库级收口（非终态置 cancelled）。
+func (m *Manager) applyCancelCommand(c ownedCommand) {
+	var p CancelCommandPayload
+	if err := json.Unmarshal([]byte(c.Payload), &p); err != nil {
+		log.Printf("[runmgr] cancel %s payload: %v", c.CommandID, err)
+		return
+	}
+	runID := c.RunID
+	if runID == "" {
+		runID = p.RunID
+	}
+	// 本地活跃 → 级联取消（含 HITL 唤醒）；否则库级收口。CancelRun 对
+	// 非本地 run 不落库（只动内存句柄），不能作为成败判据。
+	if m.RunActiveLocally(runID) {
+		if err := m.CancelRun(runID); err == nil {
+			return
+		}
+	}
+	// 库级取消（幂等——仅非终态可置）。
+	if _, err := m.st.DB().Exec(
+		`UPDATE runs SET status='cancelled', ended_at=$2 WHERE id=$1 AND status IN ('pending','running','waiting_hitl')`, runID, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
+		log.Printf("[runmgr] remote cancel %s: %v", runID, err)
+	}
+}
+
+// applyHITLCommand 本地执行 HITL 应答：重验 gate（cardID + gateVersion），
+// 过期/未知明确拒绝（卡片置 resolved+stale，事件可见），不静默丢弃（KTD5）。
+func (m *Manager) applyHITLCommand(c ownedCommand) {
+	var p HITLCommandPayload
+	if err := json.Unmarshal([]byte(c.Payload), &p); err != nil {
+		log.Printf("[runmgr] hitl %s payload: %v", c.CommandID, err)
+		return
+	}
+	resp := core.HITLResponse{CardID: core.ID(p.CardID)}
+	if p.Kind == "free" {
+		resp.FreeText = string(p.Answer)
+	} else {
+		resp.ChoiceID = p.Kind // choiceId 即按钮语义（与 cardAction 契约一致）
+	}
+	_, _, err := m.ResolveHITL(p.Owner, p.CardID, resp)
+	if err != nil {
+		// 过期/未知 gate：明确销项（resolved + 注记），发出可见结论。
+		_ = m.st.ResolveHITLCard(p.Owner, p.CardID, core.HITLResolvedPayload{
+			CardID: core.ID(p.CardID), Kind: "expired", Answer: "跨实例审批迟到：gate 已过期或执行已移交",
+		})
+		log.Printf("[runmgr] hitl %s (card %s) rejected: %v", c.CommandID, p.CardID, err)
+	}
+}

internal/runmgr/multi_instance_test.go

@@ -0,0 +1,176 @@
+package runmgr
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"agent-system-backend/internal/store"
+	core "agent-system-core"
+)
+
+// 协作存储 U5：双实例语义（KTD4/KTD5）。两个独立 Manager（A/B）共享同一
+// PG schema（等价两真实进程的存储视图；进程级隔离由 CI 双进程验收与部署
+// 合同覆盖——本测试验证所有权/围栏/命令的库级正确性）。
+// 场景：同会话争抢、跨实例取消、旧 owner 迟到结算被拒。
+
+func pgSchemaStore(t *testing.T) *store.Store {
+	t.Helper()
+	dsn := os.Getenv("AGENT_PG_TEST_DSN")
+	if dsn == "" {
+		t.Skip("AGENT_PG_TEST_DSN 未设置：真库多实例语义由 CI 必跑 job 覆盖")
+	}
+	schema := "u5_" + strings.ToLower(store.NewID("t")) + "_" + fmt.Sprint(os.Getpid())
+	admin, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
+		admin.Close()
+		t.Fatalf("pg_trgm: %v", err)
+	}
+	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
+		admin.Close()
+		t.Fatalf("schema: %v", err)
+	}
+	t.Cleanup(func() {
+		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
+		admin.Close()
+	})
+	st, err := store.OpenPostgres(store.AppendPGRuntimeParam(dsn, "search_path", schema+",public"), filepath.Join(t.TempDir(), "art"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { st.Close() })
+	return st
+}
+
+func newOwnedManager(t *testing.T, st *store.Store, tag string) *Manager {
+	t.Helper()
+	mgr := New(st, nil, nil, nil, nil, noopRunner{}, noopRunner{}, noopRunner{}, noopRunner{})
+	mgr.SetInstanceTag(tag, "127.0.0.1:0")
+	t.Cleanup(mgr.Shutdown)
+	return mgr
+}
+
+// 同会话争抢：A 占用后 B 的 ClaimConversation 被 409 拒（库级租约裁决）。
+func TestMultiInstanceConversationContention(t *testing.T) {
+	st := pgSchemaStore(t)
+	a := newOwnedManager(t, st, "instA")
+	b := newOwnedManager(t, st, "instB")
+	if err := a.ClaimConversation("cnv_race"); err != nil {
+		t.Fatalf("A claim: %v", err)
+	}
+	defer a.ReleaseConversation("cnv_race")
+	if err := b.ClaimConversation("cnv_race"); err == nil || !errors.Is(err, ErrConversationBusy) {
+		t.Fatalf("B claim should be busy: %v", err)
+	}
+	// A 释放后 B 可占。
+	a.ReleaseConversation("cnv_race")
+	if err := b.ClaimConversation("cnv_race"); err != nil {
+		t.Fatalf("B claim after release: %v", err)
+	}
+}
+
+// 旧 owner 迟到结算被拒（fencing）：A 持 epoch N；租约到期 B 接管（N+1）；
+// A 的 SettleRunFenced 被围栏拒绝、run 不变；B 的结算成功。
+func TestMultiInstanceStaleOwnerSettleFenced(t *testing.T) {
+	st := pgSchemaStore(t)
+	conv := recoveryConversation(t, st)
+	// A 以 1 秒租约占用并开 run。
+	eA, err := st.TryClaimExecution(string(conv.ID), "instA", "", 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	run := recoveryRun(t, st, conv, core.RunRunning)
+	time.Sleep(1100 * time.Millisecond) // 租约到期
+	// B 接管。
+	eB, err := st.TryClaimExecution(string(conv.ID), "instB", "", 60)
+	if err != nil || eB != eA+1 {
+		t.Fatalf("B takeover: epoch=%d (want %d) err=%v", eB, eA+1, err)
+	}
+	// A 迟到结算：fenced，run 不变（仍 running）。
+	runA := run
+	runA.Status = core.RunCompleted
+	updated, fenced, err := st.SettleRunFenced(&runA, nil, "instA", eA)
+	if err != nil || updated || !fenced {
+		t.Fatalf("A settle: updated=%v fenced=%v err=%v（应 fenced 拒写）", updated, fenced, err)
+	}
+	after, _ := st.GetRunByID(run.ID)
+	if after.Status != core.RunRunning {
+		t.Fatalf("fenced settle 不得改写 run：status=%s", after.Status)
+	}
+	// B 结算成功。
+	runB := run
+	runB.Status = core.RunCompleted
+	updated, fenced, err = st.SettleRunFenced(&runB, nil, "instB", eB)
+	if err != nil || !updated || fenced {
+		t.Fatalf("B settle: updated=%v fenced=%v err=%v", updated, fenced, err)
+	}
+}
+
+// 跨实例取消（KTD5）：run 由 A 执行；B 侧投递取消命令；A 的消费循环领取并
+// 取消（本地句柄路径）；命令恰被消费一次。
+func TestMultiInstanceCrossCancelViaCommand(t *testing.T) {
+	st := pgSchemaStore(t)
+	a := newOwnedManager(t, st, "instA")
+	conv := recoveryConversation(t, st)
+	if err := a.ClaimConversation(string(conv.ID)); err != nil {
+		t.Fatal(err)
+	}
+	defer a.ReleaseConversation(string(conv.ID))
+	run := recoveryRun(t, st, conv, core.RunRunning)
+	a.execMu.Lock()
+	a.execEpoch[string(conv.ID)], _ = st.TryClaimExecution(string(conv.ID), "instA", run.ID, 60)
+	a.execMu.Unlock()
+
+	// B 侧投递取消命令（等价 server 的跨实例路径）。
+	cmdID := store.NewID("cmd")
+	if err := st.EnqueueCommand(cmdID, "cancel", run.ID, string(conv.ID), 0,
+		CancelCommandPayload{Owner: conv.Owner, RunID: run.ID}); err != nil {
+		t.Fatal(err)
+	}
+	// A 的单轮消费：认领并执行（run 非本实例活跃 → 库级取消收口）。
+	a.consumeCommandsOnce()
+	state, _, err := st.CommandState(cmdID)
+	if err != nil || state != "consumed" {
+		t.Fatalf("command state=%s err=%v（应 consumed）", state, err)
+	}
+	after, _ := st.GetRunByID(run.ID)
+	if after.Status != core.RunCancelled {
+		t.Fatalf("run status=%s（应 cancelled）", after.Status)
+	}
+	// 重复消费不重复执行（命令已 consumed，单轮无动作）。
+	a.consumeCommandsOnce()
+	after2, _ := st.GetRunByID(run.ID)
+	if after2.Status != core.RunCancelled {
+		t.Fatalf("重复消费后 status=%s", after2.Status)
+	}
+}
+
+// 非执行实例不吃命令：命令的会话执行权在 B，A 的消费循环应跳过。
+func TestCommandOnlyConsumedByOwner(t *testing.T) {
+	st := pgSchemaStore(t)
+	a := newOwnedManager(t, st, "instA")
+	conv := recoveryConversation(t, st)
+	run := recoveryRun(t, st, conv, core.RunRunning)
+	if _, err := st.TryClaimExecution(string(conv.ID), "instB", run.ID, 60); err != nil {
+		t.Fatal(err)
+	}
+	cmdID := store.NewID("cmd")
+	if err := st.EnqueueCommand(cmdID, "cancel", run.ID, string(conv.ID), 0, nil); err != nil {
+		t.Fatal(err)
+	}
+	a.consumeCommandsOnce()
+	state, _, _ := st.CommandState(cmdID)
+	if state != "pending" {
+		t.Fatalf("非执行实例不得消费命令：state=%s", state)
+	}
+	_ = context.Background()
+}

internal/runmgr/runmgr.go

@@ -68,9 +68,13 @@ type Manager struct {
 	active map[string]*runHandle
 
 	// 会话内串行（设计并发策略）：一个会话同时至多一个在途 conversation run。
-	convMu     sync.Mutex
-	convActive map[string]bool // conversationID → 有在途 run
-	adm        *admissionGate  // 执行许可门（U3：预算默认关闭时零开销）
+	convMu      sync.Mutex
+	convActive  map[string]bool  // conversationID → 有在途 run
+	adm         *admissionGate   // 执行许可门（U3：预算默认关闭时零开销）
+	instanceTag string           // 执行身份（U5：多实例互异；空=runmgr:addr）
+	addr        string           // 监听地址（身份退化用）
+	execMu      sync.Mutex       // 执行所有权（U5：DB 租约的本地代次登记）
+	execEpoch   map[string]int64 // conversationID → 本实例当前 epoch
 
 	instanceID   string
 	shuttingDown atomic.Bool
@@ -123,7 +127,7 @@ func New(st *store.Store, b *bus.Bus, reg *tools.Registry, p *profiles.Service,
 	return &Manager{
 		st: st, bus: b, registry: reg, profiles: p, settings: s,
 		runner: runner, subRunner: subRunner, mock: mockRunner, mockSub: mockSubRunner,
-		active: map[string]*runHandle{}, convActive: map[string]bool{},
+		active: map[string]*runHandle{}, convActive: map[string]bool{}, execEpoch: map[string]int64{},
 		notices: map[string][]bgNotice{}, bgInFlight: map[string]int{}, delivering: map[string]bool{},
 		events: newEventBuffer(st, b),
 		adm:    newAdmissionGate(nil, 0),
@@ -1136,7 +1140,39 @@ func workflowScriptBodies(reg *workflow.Registry, def core.WorkflowDef) map[stri
 
 // ClaimConversation 抢占会话槽位。忙则返回 ErrConversationBusy，此时调用方不应写消息。
 func (m *Manager) ClaimConversation(conversationID string) error {
-	return m.claimConversation(conversationID)
+	if err := m.claimConversation(conversationID); err != nil {
+		return err
+	}
+	// 执行所有权（U5/KTD4）：PG 档在内存槽位之上叠加 DB 租约——跨实例的
+	// 同会话争抢由库裁决；接管递增 epoch 供结算围栏。失败时回退内存槽位。
+	if m.st.Dialect() == "postgres" {
+		epoch, err := m.st.TryClaimExecution(conversationID, m.InstanceID(), "", 15*60)
+		if err != nil {
+			m.releaseConversation(conversationID)
+			if errors.Is(err, store.ErrExecutionBusy) {
+				return ErrConversationBusy
+			}
+			return err
+		}
+		m.execMu.Lock()
+		m.execEpoch[conversationID] = epoch
+		m.execMu.Unlock()
+	}
+	return nil
+}
+
+// InstanceID 本进程执行身份（组合根可经 SetInstanceTag 覆盖；多实例必须互异）。
+func (m *Manager) InstanceID() string {
+	if m.instanceTag != "" {
+		return m.instanceTag
+	}
+	return "runmgr:" + m.addr
+}
+
+// SetInstanceTag 注入稳定实例身份与监听地址（组合根）。
+func (m *Manager) SetInstanceTag(tag, addr string) {
+	m.instanceTag = tag
+	m.addr = addr
 }
 
 // ReleaseConversation 释放已抢到的槽位。启动失败时调用；重复释放无效果。
@@ -1224,6 +1260,17 @@ func (m *Manager) releaseConversation(conversationID string) {
 	m.convMu.Lock()
 	delete(m.convActive, conversationID)
 	m.convMu.Unlock()
+	if m.st != nil && m.st.Dialect() == "postgres" {
+		m.execMu.Lock()
+		epoch, ok := m.execEpoch[conversationID]
+		if ok {
+			delete(m.execEpoch, conversationID)
+		}
+		m.execMu.Unlock()
+		if ok {
+			m.st.ReleaseExecution(conversationID, m.InstanceID(), epoch)
+		}
+	}
 }
 
 // execute 运行至终态并收敛：状态、usage、助手消息、收尾事件。
@@ -1295,12 +1342,19 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
 			msg.VariantIndex = index
 		}
 	}
-	updated, err := m.st.SettleRun(&rc.run, msg)
+	updated, fenced, err := m.settleRunWithFence(rc, msg)
 	if err != nil {
 		log.Printf("[runmgr] settle run %s: %v", rc.run.ID, err)
 		m.failSettle(rc, err)
 		return
 	}
+	if fenced {
+		// 旧执行代次（U5/KTD4）：租约已被他实例接管——本地不写终态；
+		// run 归属已移交，发围栏事件后静默退出（迟到副作用由此可见）。
+		log.Printf("[runmgr] settle run %s fenced: execution taken over (conv %s)", rc.run.ID, rc.run.ConversationID)
+		m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunFailed, core.RunFailedPayload{Code: "run.fenced", Message: "执行权已被其他实例接管"})
+		return
+	}
 	if m.OnUsage != nil && usage.TotalTokens > 0 {
 		m.OnUsage(rc.run.Owner, usage.TotalTokens)
 	}
@@ -1393,6 +1447,39 @@ func (m *Manager) emit(conversationID, runID string, typ core.StreamEventType, p
 	return m.events.emit(conversationID, runID, typ, payload)
 }
 
+// settleRunWithFence（U5）：PG 档以本实例登记的 epoch 做围栏结算。
+func (m *Manager) settleRunWithFence(rc *runCtx, msg *core.Message) (updated, fenced bool, err error) {
+	if m.st.Dialect() != "postgres" {
+		updated, err = m.st.SettleRun(&rc.run, msg)
+		return updated, false, err
+	}
+	m.execMu.Lock()
+	epoch, ok := m.execEpoch[string(rc.run.ConversationID)]
+	m.execMu.Unlock()
+	if !ok {
+		// 未登记（旧路径起跑）：按库内当前代次校验（同实例放行）。
+		if es, gerr := m.st.GetExecution(string(rc.run.ConversationID)); gerr == nil {
+			epoch = es.Epoch
+		}
+	}
+	return m.st.SettleRunFenced(&rc.run, msg, m.InstanceID(), epoch)
+}
+
+// RunActiveLocally run 是否在本实例活跃执行（U5：跨实例命令路由判定）。
+func (m *Manager) RunActiveLocally(runID string) bool {
+	root := runID
+	if r, err := m.st.GetRunByID(runID); err == nil && r.ParentRunID != "" {
+		root = r.ParentRunID
+	}
+	m.mu.Lock()
+	_, ok := m.active[root]
+	if !ok {
+		_, ok = m.active[runID]
+	}
+	m.mu.Unlock()
+	return ok
+}
+
 // CancelRun 取消 run 及其子 run（ctx 级联）。
 func (m *Manager) CancelRun(runID string) error {
 	r, err := m.st.GetRunByID(runID)

internal/server/server.go

@@ -1006,7 +1006,24 @@ func (s *Server) listAudits(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) cancelRun(w http.ResponseWriter, r *http.Request) {
-	writeOKOr(w, s.mgr.CancelRun(r.PathValue("id")))
+	runID := r.PathValue("id")
+	err := s.mgr.CancelRun(runID)
+	if err == nil {
+		writeOKOr(w, nil)
+		return
+	}
+	// 跨实例取消（协作存储 U5/KTD5）：run 不在本实例执行（无句柄）且 PG 档
+	// ——写持久命令，由执行实例消费；API 立即受理（命令可见、唯一消费）。
+	if s.st.Dialect() == "postgres" && s.mgr.RunActiveLocally(runID) == false {
+		if cerr := s.st.EnqueueCommand(store.NewID("cmd"), "cancel", runID, "", 0,
+			runmgr.CancelCommandPayload{Owner: s.owner(r), RunID: runID}); cerr != nil {
+			writeStoreErr(w, err)
+			return
+		}
+		writeJSON(w, http.StatusAccepted, map[string]any{"accepted": true, "via": "command"})
+		return
+	}
+	writeStoreErr(w, err)
 }
 
 func (s *Server) getRun(w http.ResponseWriter, r *http.Request) {
@@ -1029,6 +1046,21 @@ func (s *Server) resolveHITL(w http.ResponseWriter, r *http.Request) {
 	}
 	resp.CardID = r.PathValue("id")
 	pivotText, pivoted, err := s.mgr.ResolveHITL(s.owner(r), string(resp.CardID), resp)
+	if err != nil && s.st.Dialect() == "postgres" {
+		// 跨实例 HITL（U5/KTD5）：本地无执行句柄——写持久命令，执行实例
+		// 消费时重验 gate（cardID+gateVersion），过期审批明确拒绝。
+		if card, gerr := s.st.GetHITLCard(s.owner(r), string(resp.CardID)); gerr == nil && !card.Resolved {
+			kind := resp.ChoiceID
+			if kind == "" {
+				kind = "free"
+			}
+			if cerr := s.st.EnqueueCommand(store.NewID("cmd"), "hitl", string(card.RunID), string(card.ConversationID), 0,
+				runmgr.HITLCommandPayload{Owner: s.owner(r), CardID: string(resp.CardID), Kind: kind, Answer: json.RawMessage(resp.FreeText)}); cerr == nil {
+				writeJSON(w, http.StatusAccepted, map[string]any{"accepted": true, "via": "command"})
+				return
+			}
+		}
+	}
 	if err != nil {
 		writeStoreErr(w, err)
 		return

internal/store/execution.go

@@ -0,0 +1,219 @@
+package store
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+)
+
+// 协作存储 U5（KTD4/KTD5）：会话执行租约 + 持久命令收件箱。
+//
+// 租约：conversation_executions 一行一会话。TryClaimExecution 空闲/租约到期
+// 才可占用；接管（含同实例重新占用）递增 epoch——旧 owner 的续租与结算按
+//（instance, epoch）双重校验被拒（fencing）。lease_until 用应用侧 UTC ISO
+// （单库写入方一致即可；跨库时钟偏移属部署合同，见 runbook）。
+//
+// 命令：run_commands 收件箱承载跨实例取消/HITL（KTD5）。command_id 幂等；
+// pending→consumed 由消费方原子认领（PG FOR UPDATE SKIP LOCKED；SQLite 单写
+// 者条件更新）。命令不删除——审计与重放判定保留。
+
+// ErrExecutionBusy 会话执行权被其他实例持有且租约未到期。
+var ErrExecutionBusy = errors.New("store: execution owned by another instance")
+
+// ErrFenced 旧执行代次的写入被围栏拒绝（epoch 已被接管递增）。
+var ErrFenced = errors.New("store: fenced by newer execution epoch")
+
+// ExecutionState 当前执行租约快照（诊断/测试）。
+type ExecutionState struct {
+	ConversationID string
+	InstanceID     string
+	RunID          string
+	Epoch          int64
+	LeaseUntil     string
+}
+
+// TryClaimExecution 占用（或租约到期后接管）会话执行权。
+// 返回接管后的 epoch（接管必然 +1）。被他人持有且未到期 → ErrExecutionBusy。
+func (s *Store) TryClaimExecution(conversationID, instanceID, runID string, leaseSec int) (int64, error) {
+	if leaseSec <= 0 {
+		leaseSec = 90
+	}
+	tx, err := s.db.Begin()
+	if err != nil {
+		return 0, err
+	}
+	defer tx.Rollback()
+	if err := s.lockAdmission(tx); err != nil { // 复用全局账本锁串行化清点
+		return 0, err
+	}
+	now := time.Now().UTC()
+	until := now.Add(time.Duration(leaseSec) * time.Second).UTC().Format(time.RFC3339Nano)
+	var inst, lease string
+	var epoch int64
+	err = tx.QueryRow(s.q(`SELECT instance_id, epoch, lease_until FROM conversation_executions WHERE conversation_id=?`), conversationID).Scan(&inst, &epoch, &lease)
+	switch {
+	case errors.Is(err, sql.ErrNoRows):
+		if _, err := tx.Exec(s.q(`INSERT INTO conversation_executions (conversation_id, instance_id, run_id, epoch, lease_until, updated_at) VALUES (?,?,?,1,?,?)`),
+			conversationID, instanceID, runID, until, nowISO()); err != nil {
+			return 0, err
+		}
+		if err := tx.Commit(); err != nil {
+			return 0, err
+		}
+		return 1, nil
+	case err != nil:
+		return 0, err
+	}
+	if inst != instanceID && lease > nowISO() {
+		return epoch, fmt.Errorf("conv %s: %w", conversationID, ErrExecutionBusy)
+	}
+	// 接管（到期或同实例重占）：epoch+1，fencing 由此产生。
+	if _, err := tx.Exec(s.q(`UPDATE conversation_executions SET instance_id=?, run_id=?, epoch=epoch+1, lease_until=?, updated_at=? WHERE conversation_id=?`),
+		instanceID, runID, until, nowISO(), conversationID); err != nil {
+		return 0, err
+	}
+	if err := tx.Commit(); err != nil {
+		return 0, err
+	}
+	return epoch + 1, nil
+}
+
+// RenewExecution 续租（fencing：instance+epoch 双校验）。假 = 已被接管或释放。
+func (s *Store) RenewExecution(conversationID, instanceID string, epoch int64, leaseSec int) bool {
+	if leaseSec <= 0 {
+		leaseSec = 90
+	}
+	until := time.Now().UTC().Add(time.Duration(leaseSec) * time.Second).UTC().Format(time.RFC3339Nano)
+	res, err := s.db.Exec(s.q(`UPDATE conversation_executions SET lease_until=?, updated_at=? WHERE conversation_id=? AND instance_id=? AND epoch=?`),
+		until, nowISO(), conversationID, instanceID, epoch)
+	if err != nil {
+		return false
+	}
+	n, _ := res.RowsAffected()
+	return n > 0
+}
+
+// ReleaseExecution 主动释放（优雅停机）：行保留、租约置为已过期——epoch 保留，
+// 下一次占用仍递增（fencing 单调）。
+func (s *Store) ReleaseExecution(conversationID, instanceID string, epoch int64) {
+	_, _ = s.db.Exec(s.q(`UPDATE conversation_executions SET lease_until=?, updated_at=? WHERE conversation_id=? AND instance_id=? AND epoch=?`),
+		time.Now().UTC().Add(-time.Second).UTC().Format(time.RFC3339Nano), nowISO(), conversationID, instanceID, epoch)
+}
+
+// GetExecution 读取当前租约（无行 = 从未占用）。
+func (s *Store) GetExecution(conversationID string) (ExecutionState, error) {
+	var e ExecutionState
+	err := s.db.QueryRow(s.q(`SELECT conversation_id, instance_id, COALESCE(run_id,''), epoch, lease_until FROM conversation_executions WHERE conversation_id=?`), conversationID).
+		Scan(&e.ConversationID, &e.InstanceID, &e.RunID, &e.Epoch, &e.LeaseUntil)
+	if errors.Is(err, sql.ErrNoRows) {
+		return e, ErrNotFound
+	}
+	return e, err
+}
+
+// ExecutionFenceOK 校验 (instance, epoch) 仍是当前执行权（结算/事件写入的
+// 前置检查；事务内校验见 SettleRunFenced）。
+func (s *Store) ExecutionFenceOK(conversationID, instanceID string, epoch int64) bool {
+	var n int
+	if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM conversation_executions WHERE conversation_id=? AND instance_id=? AND epoch=?`), conversationID, instanceID, epoch).Scan(&n); err != nil {
+		return false
+	}
+	return n > 0
+}
+
+// SettleRunFenced 带执行围栏的终态结算（KTD4：旧 owner 恢复后不能提交）。
+// fenced=true 表示 (instance,epoch) 已不是当前执行权——run 未被改动，调用方
+// 应将本地 run 视为被接管（不写终态、事件标注接管）。无租约行（SQLite 本地
+// 档/未启用执行权）恒放行。
+
+// RunCommand 收件箱命令。
+type RunCommand struct {
+	CommandID      string
+	Kind           string // cancel | hitl
+	RunID          string
+	ConversationID string
+	Epoch          int64
+	Payload        string
+	CreatedAt      string
+}
+
+// EnqueueCommand 投递跨实例命令（幂等：同 command_id 重复投递不产生新行）。
+func (s *Store) EnqueueCommand(commandID, kind, runID, conversationID string, epoch int64, payload any) error {
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	_, err = s.db.Exec(s.q(`INSERT INTO run_commands (command_id, kind, run_id, conversation_id, epoch, payload, state, created_at)
+		VALUES (?,?,?,?,?,?,'pending',?)
+		ON CONFLICT(command_id) DO NOTHING`),
+		commandID, kind, runID, conversationID, epoch, string(raw), nowISO())
+	if err != nil {
+		return fmt.Errorf("enqueue command %s: %w", commandID, err)
+	}
+	return nil
+}
+
+// ClaimPendingCommands 认领待处理命令（原子 pending→consumed）。
+// PG：FOR UPDATE SKIP LOCKED（多实例并发认领不重复）；SQLite：单写者条件更新。
+// 只认领「本实例相关」的命令由调用方过滤——收件箱按执行权路由：命令在
+// 被执行实例消费（cancel/hitl 目标 run 的当前 owner）。
+func (s *Store) ClaimPendingCommands(consumer string, limit int) ([]RunCommand, error) {
+	if limit <= 0 {
+		limit = 16
+	}
+	tx, err := s.db.Begin()
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+	var rows *sql.Rows
+	if s.dialect.name() == "postgres" {
+		rows, err = tx.Query(`SELECT command_id, kind, run_id, conversation_id, epoch, payload, created_at
+			FROM run_commands WHERE state='pending' ORDER BY created_at
+			LIMIT $1 FOR UPDATE SKIP LOCKED`, limit)
+	} else {
+		rows, err = tx.Query(s.q(`SELECT command_id, kind, run_id, conversation_id, epoch, payload, created_at
+			FROM run_commands WHERE state='pending' ORDER BY created_at LIMIT ?`), limit)
+	}
+	if err != nil {
+		return nil, err
+	}
+	var out []RunCommand
+	for rows.Next() {
+		var c RunCommand
+		if err := rows.Scan(&c.CommandID, &c.Kind, &c.RunID, &c.ConversationID, &c.Epoch, &c.Payload, &c.CreatedAt); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		out = append(out, c)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	for _, c := range out {
+		res, err := tx.Exec(s.q(`UPDATE run_commands SET state='consumed', consumed_at=?, consumer=? WHERE command_id=? AND state='pending'`),
+			nowISO(), consumer, c.CommandID)
+		if err != nil {
+			return nil, err
+		}
+		if n, _ := res.RowsAffected(); n == 0 {
+			return nil, fmt.Errorf("command %s claimed concurrently", c.CommandID)
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// CommandState 查询单条命令状态（测试/诊断）。
+func (s *Store) CommandState(commandID string) (state, consumer string, err error) {
+	err = s.db.QueryRow(s.q(`SELECT state, COALESCE(consumer,'') FROM run_commands WHERE command_id=?`), commandID).Scan(&state, &consumer)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", "", ErrNotFound
+	}
+	return
+}

internal/store/lease_pg_test.go

@@ -0,0 +1,121 @@
+package store
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+// 协作存储 U5（KTD4）：会话执行租约与 epoch fencing。
+// 双档（PG 由 CI 真库 job 覆盖咨询锁与跨实例语义；SQLite 验证条件逻辑）。
+
+func TestExecutionClaimAndContention(t *testing.T) {
+	st := openTestStore(t)
+	// 首次占用：epoch=1。
+	e1, err := st.TryClaimExecution("cnv_1", "instA", "run_1", 60)
+	if err != nil || e1 != 1 {
+		t.Fatalf("claim A: epoch=%d err=%v", e1, err)
+	}
+	// 他实例争抢：租约未到期 → ErrExecutionBusy。
+	_, err = st.TryClaimExecution("cnv_1", "instB", "run_2", 60)
+	if !errors.Is(err, ErrExecutionBusy) {
+		t.Fatalf("B should be busy: %v", err)
+	}
+	// 同实例重占（新回合）：允许，epoch 递增。
+	e2, err := st.TryClaimExecution("cnv_1", "instA", "run_3", 60)
+	if err != nil || e2 != 2 {
+		t.Fatalf("re-claim A: epoch=%d err=%v", e2, err)
+	}
+	// 续租 fencing：正确 (instance,epoch) 成功；旧 epoch 失败。
+	if !st.RenewExecution("cnv_1", "instA", e2, 60) {
+		t.Fatal("renew with current epoch should succeed")
+	}
+	if st.RenewExecution("cnv_1", "instA", e1, 60) {
+		t.Fatal("renew with stale epoch must fail")
+	}
+	// 主动释放后他实例可占，epoch 继续递增（单调 fencing）。
+	st.ReleaseExecution("cnv_1", "instA", e2)
+	e3, err := st.TryClaimExecution("cnv_1", "instB", "run_4", 60)
+	if err != nil || e3 != 3 {
+		t.Fatalf("takeover B: epoch=%d err=%v", e3, err)
+	}
+	// fence 校验：B 当前 OK；A 的旧代次失败。
+	if !st.ExecutionFenceOK("cnv_1", "instB", e3) {
+		t.Fatal("B fence should be ok")
+	}
+	if st.ExecutionFenceOK("cnv_1", "instA", e2) {
+		t.Fatal("A stale fence must fail")
+	}
+}
+
+func TestExecutionLeaseExpiryTakeover(t *testing.T) {
+	st := openTestStore(t)
+	// 1 秒租约到期后他实例接管。
+	if _, err := st.TryClaimExecution("cnv_2", "instA", "run_1", 1); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(1100 * time.Millisecond)
+	e2, err := st.TryClaimExecution("cnv_2", "instB", "run_2", 60)
+	if err != nil || e2 != 2 {
+		t.Fatalf("expired takeover: epoch=%d err=%v", e2, err)
+	}
+	// 旧 owner 续租被拒（fencing）。
+	if st.RenewExecution("cnv_2", "instA", 1, 60) {
+		t.Fatal("stale renew must fail after takeover")
+	}
+}
+
+// 命令收件箱（KTD5）：幂等投递、原子认领、并发认领不重复。
+func TestRunCommandsClaim(t *testing.T) {
+	st := openTestStore(t)
+	if err := st.EnqueueCommand("cmd_1", "cancel", "run_9", "cnv_9", 0, map[string]string{"owner": "o"}); err != nil {
+		t.Fatal(err)
+	}
+	// 幂等：重复投递不新增。
+	if err := st.EnqueueCommand("cmd_1", "cancel", "run_9", "cnv_9", 0, nil); err != nil {
+		t.Fatal(err)
+	}
+	got, err := st.ClaimPendingCommands("instA", 8)
+	if err != nil || len(got) != 1 || got[0].CommandID != "cmd_1" || got[0].Kind != "cancel" {
+		t.Fatalf("claim: %v err=%v", got, err)
+	}
+	state, consumer, err := st.CommandState("cmd_1")
+	if err != nil || state != "consumed" || consumer != "instA" {
+		t.Fatalf("state=%s consumer=%s err=%v", state, consumer, err)
+	}
+	// 已消费不再返回。
+	got, err = st.ClaimPendingCommands("instB", 8)
+	if err != nil || len(got) != 0 {
+		t.Fatalf("consumed must not re-deliver: %v err=%v", got, err)
+	}
+}
+
+func TestRunCommandsConcurrentSingleClaim(t *testing.T) {
+	st := openTestStore(t)
+	for i := 0; i < 8; i++ {
+		if err := st.EnqueueCommand(newID("cmd"), "cancel", newID("run"), "cnv_c", 0, nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var mu = make(chan int, 2)
+	total := 0
+	done := make(chan struct{})
+	for w := 0; w < 2; w++ {
+		go func(w int) {
+			got, err := st.ClaimPendingCommands("inst"+string(rune('A'+w)), 8)
+			if err == nil {
+				mu <- len(got)
+			}
+			done <- struct{}{}
+		}(w)
+	}
+	<-done
+	<-done
+	close(mu)
+	for n := range mu {
+		total += n
+	}
+	if total != 8 {
+		t.Fatalf("并发认领应恰好 8 条（不重复不丢失）：total=%d", total)
+	}
+}

internal/store/migrate.go

@@ -592,6 +592,63 @@ func init() {
 				`DROP TABLE IF EXISTS surface_revisions`,
 				`DROP TABLE IF EXISTS surfaces`,
 			}}},
+		migration{version: 23, name: "execution_ownership_commands",
+			// 执行所有权与持久命令（协作存储 U5，KTD4/KTD5）：
+			// conversation_executions = 会话执行租约（实例/run/epoch/到期），
+			// 接管递增 epoch 作 fencing；run_commands = 跨实例命令收件箱
+			//（取消/HITL：command_id 幂等、state pending/consumed）。
+			up: map[string][]string{"sqlite": {
+				`CREATE TABLE IF NOT EXISTS conversation_executions (
+					conversation_id TEXT PRIMARY KEY,
+					instance_id TEXT NOT NULL,
+					run_id TEXT NOT NULL DEFAULT '',
+					epoch INTEGER NOT NULL DEFAULT 0,
+					lease_until TEXT NOT NULL,
+					updated_at TEXT NOT NULL
+				)`,
+				`CREATE TABLE IF NOT EXISTS run_commands (
+					command_id TEXT PRIMARY KEY,
+					kind TEXT NOT NULL,
+					run_id TEXT NOT NULL DEFAULT '',
+					conversation_id TEXT NOT NULL DEFAULT '',
+					epoch INTEGER NOT NULL DEFAULT 0,
+					payload TEXT NOT NULL DEFAULT '{}',
+					state TEXT NOT NULL DEFAULT 'pending',
+					created_at TEXT NOT NULL,
+					consumed_at TEXT,
+					consumer TEXT NOT NULL DEFAULT ''
+				)`,
+				`CREATE INDEX IF NOT EXISTS idx_run_commands_pending ON run_commands(state, created_at)`,
+			}, "postgres": {
+				`CREATE TABLE IF NOT EXISTS conversation_executions (
+					conversation_id TEXT PRIMARY KEY,
+					instance_id TEXT NOT NULL,
+					run_id TEXT NOT NULL DEFAULT '',
+					epoch INTEGER NOT NULL DEFAULT 0,
+					lease_until TEXT NOT NULL,
+					updated_at TEXT NOT NULL
+				)`,
+				`CREATE TABLE IF NOT EXISTS run_commands (
+					command_id TEXT PRIMARY KEY,
+					kind TEXT NOT NULL,
+					run_id TEXT NOT NULL DEFAULT '',
+					conversation_id TEXT NOT NULL DEFAULT '',
+					epoch INTEGER NOT NULL DEFAULT 0,
+					payload TEXT NOT NULL DEFAULT '{}',
+					state TEXT NOT NULL DEFAULT 'pending',
+					created_at TEXT NOT NULL,
+					consumed_at TEXT,
+					consumer TEXT NOT NULL DEFAULT ''
+				)`,
+				`CREATE INDEX IF NOT EXISTS idx_run_commands_pending ON run_commands(state, created_at)`,
+			}},
+			down: map[string][]string{"sqlite": {
+				`DROP TABLE IF EXISTS run_commands`,
+				`DROP TABLE IF EXISTS conversation_executions`,
+			}, "postgres": {
+				`DROP TABLE IF EXISTS run_commands`,
+				`DROP TABLE IF EXISTS conversation_executions`,
+			}}},
 	)
 }
 

internal/store/recovery.go

@@ -122,6 +122,42 @@ func (s *Store) ActiveHITLCardsByRun(runID string) ([]core.HITLCard, error) {
 // SettleRun 在一个事务里更新 run，并可选地追加助手消息。
 // updated 为假表示条件更新没有改到行（已被收成终态）；消息仍会写入。
 // 序号冲突时整段事务重试。其它失败回滚，run 保持原状态。
+// SettleRunFenced（协作存储 U5/KTD4）：终态结算带执行围栏——事务内校验
+// (instance,epoch) 仍是该会话的当前执行权，被接管则拒写（fenced=true，run
+// 不变）。无租约行（未启用执行所有权）恒放行，语义与 SettleRun 一致。
+func (s *Store) SettleRunFenced(r *core.Run, msg *core.Message, instanceID string, epoch int64) (updated, fenced bool, err error) {
+	s.settleMu.Lock()
+	fault := s.settleFault
+	s.settleMu.Unlock()
+	if fault != nil {
+		return false, false, fault
+	}
+	if msg != nil && msg.ID == "" {
+		msg.ID = newID("msg")
+	}
+	var last error
+	for attempt := 0; attempt < 8; attempt++ {
+		updated, fenced, last = s.settleRunOnce(r, msg, &fenceRef{instance: instanceID, epoch: epoch, conversation: string(r.ConversationID)})
+		if last == nil {
+			return updated, fenced, nil
+		}
+		if !isUniqueViolation(last) {
+			return false, 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, false, last
+}
+
+// fenceRef 执行围栏引用（nil = 不校验）。
+type fenceRef struct {
+	instance, conversation string
+	epoch                  int64
+}
+
 func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err error) {
 	s.settleMu.Lock()
 	fault := s.settleFault
@@ -134,9 +170,10 @@ func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err err
 	}
 	var last error
 	for attempt := 0; attempt < 8; attempt++ {
-		updated, last = s.settleRunOnce(r, msg)
+		var fenced bool
+		updated, fenced, last = s.settleRunOnce(r, msg, nil)
 		if last == nil {
-			return updated, nil
+			return updated && !fenced, nil
 		}
 		if !isUniqueViolation(last) {
 			return false, last
@@ -149,14 +186,31 @@ func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err err
 	return false, last
 }
 
-func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) {
+func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef) (bool, bool, error) {
 	tx, err := s.db.Begin()
 	if err != nil {
-		return false, err
+		return false, false, err
 	}
 	defer tx.Rollback()
 	if err := s.dialect.lockConversation(tx, string(r.ConversationID)); err != nil {
-		return false, err
+		return false, false, err
+	}
+	// 执行围栏（KTD4）：会话存在租约行且 (instance,epoch) 不匹配 → 拒写。
+	// 无行 = 该会话未启用执行所有权（本地档/旧路径），放行。
+	if fence != nil {
+		var n int
+		if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM conversation_executions WHERE conversation_id=?`), fence.conversation).Scan(&n); err != nil {
+			return false, false, err
+		}
+		if n > 0 {
+			var m int
+			if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM conversation_executions WHERE conversation_id=? AND instance_id=? AND epoch=?`), fence.conversation, fence.instance, fence.epoch).Scan(&m); err != nil {
+				return false, false, err
+			}
+			if m == 0 {
+				return false, true, nil // fenced：旧执行代次，run 不变
+			}
+		}
 	}
 	var ended any
 	if r.EndedAt != nil {
@@ -165,17 +219,17 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) {
 	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
+		return false, false, err
 	}
 	n, err := res.RowsAffected()
 	if err != nil {
-		return false, err
+		return false, 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
+			return false, false, err
 		}
 		if msg.VariantGroup != "" || msg.Excluded {
 			ex := 0
@@ -183,17 +237,17 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) {
 				ex = 1
 			}
 			if _, err := tx.Exec(s.q(`UPDATE messages SET excluded=?, variant_group=?, variant_index=? WHERE id=?`), ex, msg.VariantGroup, msg.VariantIndex, msg.ID); err != nil {
-				return false, err
+				return false, false, err
 			}
 		}
 	}
 	if err := tx.Commit(); err != nil {
-		return false, err
+		return false, false, err
 	}
 	if msg != nil {
 		s.TouchConversation(string(msg.ConversationID))
 	}
-	return n == 1, nil
+	return n == 1, false, nil
 }
 
 // ListInterruptedSubagentRuns 列出被收成 interrupted 的子代理 run