package daemon

// Real committed-DB full-chain fence test: the actual TiDB-backed workflowrun
// Store, a real daemon.Hub with a registered connected client, the production
// HubWorkingReader, and the RecoveryRunner are wired together end to end. The
// store's narrow TestRecoveryTickBarrier hook holds the effect/Run rows after
// FOR UPDATE and before commit, so RecoveryTick has acquired the observation
// lease but is blocked at the row-locked boundary. Each authoritative Hub
// writer (incremental turn update, ready snapshot/running-set replacement,
// transport session-generation change, reconnect/register replacement, and
// disconnect) must then be unable to cross the machine observation guard write
// side until the tick commits or rolls back; the hook then releases, the real
// tick commits/rolls back, the runner releases the lease, and the writer
// proceeds. Assertions check the exact durable park/recovery state and the
// writer's landed effect.

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/agentruntime"
	"github.com/tidbcloud/AgentTeam/internal/workflow/orchestrator"
)

// fullChainDBFixture holds the durable identifiers of one seeded awaiting
// callback Run + effect inside the committed TiDB database.
type fullChainDBFixture struct {
	WorkspaceID string
	GroupID     string
	RunID       string
	MemberID    string
	EffectID    string
	AgentID     string
	MachineID   string
	Now         time.Time
}

// seedFullChainDBRun inserts a running Run + awaiting send_message.callback
// effect with a frozen responder into the real database, mirroring the store
// integration fixture SQL. The Run starts queued (like the production reducer)
// and is flipped to running, which is the state the recovery scan expects.
func seedFullChainDBRun(t *testing.T, pool coredb.QueryExecer, suffix string) fullChainDBFixture {
	t.Helper()
	now := time.Date(2026, 7, 21, 13, 0, 0, 123456000, time.UTC)
	fixture := fullChainDBFixture{
		WorkspaceID: "wsp_fc_" + suffix,
		GroupID:     "group_fc_" + suffix,
		RunID:       "wri_fc_" + suffix,
		MemberID:    "mem_fc_" + suffix,
		EffectID:    "wfx_fc_" + suffix,
		AgentID:     "agt_fc_" + suffix,
		MachineID:   "mach_fc_" + suffix,
		Now:         now,
	}
	userID, orgID := "usr_fc_"+suffix, "org_fc_"+suffix
	invocationID := "wiv_fc_" + suffix
	input := []byte(`{"seed":"start"}`)
	mustExecDaemonTest(t, pool, `INSERT INTO users (id,email,name,display_name,password_hash,created_at,updated_at) VALUES ($1,$2,$3,$3,'hash',$4,$4)`, userID, suffix+"@example.com", "fc-"+suffix, now)
	mustExecDaemonTest(t, pool, `INSERT INTO organizations (id,slug,name,created_by_user_id,created_at,updated_at) VALUES ($1,$2,$2,$3,$4,$4)`, orgID, "fc-"+suffix, userID, now)
	mustExecDaemonTest(t, pool, `INSERT INTO workspaces (id,organization_id,slug,name,created_at,updated_at) VALUES ($1,$2,$3,$3,$4,$4)`, fixture.WorkspaceID, orgID, "fc-"+suffix, now)
	mustExecDaemonTest(t, pool, `INSERT INTO members (id,workspace_id,type,display_name,user_id,workspace_role,status,created_at,updated_at) VALUES ($1,$2,'human',$3,$4,'member','active',$5,$5)`, fixture.MemberID, fixture.WorkspaceID, "fc-"+suffix, userID, now)
	mustExecDaemonTest(t, pool, `INSERT INTO workspace_placements (workspace_id,workspace_group_id,state,placement_epoch,created_at,updated_at) VALUES ($1,$2,'active',1,$3,$3)`, fixture.WorkspaceID, fixture.GroupID, now)
	mustExecDaemonTest(t, pool, `
INSERT INTO workflow_invocations (
 id,workspace_id,binding_id,binding_revision_id,binding_revision_number,binding_source_id,binding_activation_number,
 plan_id,plan_version_id,workflow_name_snapshot,source_kind,source_occurrence_ref,source_occurred_at,
 input_json,input_digest,source_provenance_json,
 owner_user_id_snapshot,owner_member_id_snapshot,principal_member_id_snapshot,author_member_id_snapshot,
 invoked_by_member_id,invoked_by_actor_type,invoked_by_actor_id,idempotency_key,request_digest,accepted_at
) VALUES (
	$1,$2,NULL,NULL,NULL,NULL,NULL,$3,$4,$13,'manual',$5,$6,
	$7,$8,'{}',$9,$10,$10,$10,$10,'human',$9,$11,$12,$6
)`, invocationID, fixture.WorkspaceID, "pln_fc_"+suffix, "plv_fc_"+suffix, "manual:"+suffix, now,
		input, fmt.Sprintf("%064x", 7), userID, fixture.MemberID, "accept-"+suffix, fmt.Sprintf("%064x", 42), "FullChain "+suffix)
	mustExecDaemonTest(t, pool, `
INSERT INTO workflow_run_instances (
	 id,workspace_id,invocation_id,plan_id,plan_version_id,workflow_name_snapshot,status,
	 input_json,input_digest,state_json,
	 owner_user_id_snapshot,owner_member_id_snapshot,principal_member_id_snapshot,author_member_id_snapshot,
	 queued_at,revision,last_event_seq,runnable_at,created_at,updated_at
) VALUES (
	$1,$2,$3,$4,$5,$11,'running',$6,$7,'{}',$8,$9,$9,$9,$10,1,1,$10,$10,$10
)`, fixture.RunID, fixture.WorkspaceID, invocationID,
		"pln_fc_"+suffix, "plv_fc_"+suffix, input, fmt.Sprintf("%064x", 7), userID, fixture.MemberID, now, "FullChain "+suffix)
	mustExecDaemonTest(t, pool, `
INSERT INTO workflow_run_instance_events (id,workspace_id,run_id,seq,event_type,payload_json,created_at)
VALUES ($1,$2,$3,1,'run_accepted','{"status":"running"}',$4)`, "wre_fc_"+suffix, fixture.WorkspaceID, fixture.RunID, now)

	activationID := "wra_" + fixture.EffectID
	requestJSON := fmt.Sprintf(`{"intent":{"activationId":"%s","nodeId":"callback","kind":"send_message","callback":true,"responder":"%s","retry":{"maxAttempts":1,"backoffNanos":1000000000}},"message":{"workspaceId":"%s","planId":"pln_fc","planVersionId":"plv_fc","runId":"%s","activationId":"%s","effectId":"%s","nodeId":"callback","sourceKind":"manual","ownerUserId":"usr_fc","authorMemberId":"%s","clientMessageId":"workflow-run:%s","content":"answer","target":"#fc","callback":true,"responderAgentRef":"%s"}}`,
		activationID, fixture.AgentID, fixture.WorkspaceID, fixture.RunID, activationID, fixture.EffectID, fixture.MemberID, fixture.EffectID, fixture.AgentID)
	receipt, err := json.Marshal(map[string]any{
		"messageId":                "msg_" + fixture.EffectID,
		"wakeAgentIds":             []string{fixture.AgentID},
		"callbackResponderAgentId": fixture.AgentID,
		"deliveryScope":            map[string]any{"kind": "channel", "scopeId": "chn_fc"},
	})
	require.NoError(t, err)
	mustExecDaemonTest(t, pool, `
INSERT INTO workflow_run_activations (
 id,workspace_id,run_id,logical_key,parent_activation_id,node_id,opcode,status,
 input_json,output_json,scope_context_json,selected_port,attempt_count,retry_at,wake_at,deadline_at,
 error_code,error_json,started_at,completed_at,created_at,updated_at
) VALUES ($1,$2,$3,'cb/fc',NULL,'callback','send_message.callback','waiting',
 '{}',NULL,'{}',NULL,1,NULL,NULL,NULL,NULL,NULL,NULL,NULL,$4,$4)
ON DUPLICATE KEY UPDATE node_id=node_id`, activationID, fixture.WorkspaceID, fixture.RunID, now)
	mustExecDaemonTest(t, pool, `
INSERT INTO scopes (id,workspace_id,type,identity_key,status,created_by_member_id,created_at,updated_at)
VALUES ($1,$2,'channel','ch:fc','active',$3,$4,$4)
ON DUPLICATE KEY UPDATE identity_key=identity_key`, "scp_fc_"+fixture.EffectID, fixture.WorkspaceID, fixture.MemberID, now)
	mustExecDaemonTest(t, pool, `
INSERT INTO scope_messages (id,workspace_id,scope_id,seq,workspace_global_seq,author_member_id,author_type,client_message_id,idempotency_payload_digest,message_type,content,created_at,updated_at)
VALUES ($1,$2,$3,1,1,$4,'human',$5,$6,'text','callback',$7,$7)
ON DUPLICATE KEY UPDATE message_type=message_type`, "msg_"+fixture.EffectID, fixture.WorkspaceID, "scp_fc_"+fixture.EffectID, fixture.MemberID, "workflow-run:rec:"+fixture.EffectID, "0000000000000000000000000000000000000000000000000000000000000099", now)
	mustExecDaemonTest(t, pool, `
INSERT INTO workflow_run_effects (
 id,workspace_id,run_id,activation_id,callback_responder_agent_id,message_id,effect_kind,idempotency_key,status,
 request_json,result_json,attempt_count,retry_at,deadline_at,lease_owner,lease_token,lease_expires_at,
 error_code,error_json,delivered_at,completed_at,created_at,updated_at
) VALUES ($1,$2,$3,$4,$5,'msg_`+fixture.EffectID+`','send_message','cb/fc','awaiting',
 $6,$7,0,NULL,NULL,NULL,NULL,NULL,NULL,NULL,$8,NULL,$8,$8)`,
		fixture.EffectID, fixture.WorkspaceID, fixture.RunID, activationID, fixture.AgentID, requestJSON, string(receipt), now)
	return fixture
}

// newFullChainDBHub builds a real Hub with one connected client for the fixture
// machine whose current connection is idle (the parked responder). The client is
// registered through the production register path so the guard/lifecycle wiring
// is real; the websocket conn is a real pair so a reconnect replacement can
// close the superseded connection through the production close path.
func newFullChainDBHub(t *testing.T, fixture fullChainDBFixture, idle bool) (*Hub, *clientState) {
	t.Helper()
	hub := &Hub{clients: map[string]*clientState{}}
	client := &clientState{
		machineID:             fixture.MachineID,
		workspaceID:           fixture.WorkspaceID,
		conn:                  newProtocolTestWebSocketPair(t),
		sessionGeneration:     "sess_fc_1",
		runningAgents:         map[string]bool{fixture.AgentID: true},
		startingAgents:        map[string]bool{},
		stoppedAgents:         map[string]bool{},
		agentStatuses:         map[string]string{},
		agentActivities:       map[string]agentActivityState{},
		agentSessions:         map[string]string{fixture.AgentID: "sess_fc_1"},
		agentTurnStateRecords: map[string]agentTurnStateRecord{fixture.AgentID: {State: "idle", Revision: 7, Generation: "sess_fc_1", SessionID: "sess_fc_1"}},
		turnStateHighWater:    7,
		connectedAt:           time.Now().UTC(),
	}
	if !idle {
		client.agentTurnStateRecords[fixture.AgentID] = agentTurnStateRecord{State: "active", Revision: 8, Generation: "sess_fc_1", SessionID: "sess_fc_1"}
	}
	client.guard = hub.observationGuard(fixture.MachineID)
	hub.register(client)
	return hub, client
}

// fullChainDBResolver resolves the fixture machine for every responder.
type fullChainDBResolver struct{ fixture fullChainDBFixture }

func (r fullChainDBResolver) MachineIDForAgent(context.Context, string, string) (string, error) {
	return r.fixture.MachineID, nil
}

// fullChainDBWriter is a real write-recording message writer for the runner.
type fullChainDBWriter struct {
	mu     sync.Mutex
	writes int
}

func (w *fullChainDBWriter) WriteRunMessage(context.Context, orchestrator.MessageRequest) (orchestrator.MessageResult, error) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.writes++
	return orchestrator.MessageResult{}, nil
}

// dbBarrier is the narrow Store test hook: it signals entered once the
// RecoveryTick transaction has acquired the effect/Run row locks (FOR UPDATE),
// then blocks until released or the context is canceled. It optionally forces a
// DB error to prove the rollback path releases the observation lease.
type dbBarrier struct {
	entered chan struct{}
	release chan struct{}
	err     error
	once    sync.Once
}

func newDBBarrier() *dbBarrier {
	return &dbBarrier{entered: make(chan struct{}), release: make(chan struct{})}
}

func (b *dbBarrier) hook(ctx context.Context) error {
	b.once.Do(func() { close(b.entered) })
	select {
	case <-b.release:
		return b.err
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (b *dbBarrier) unblock() { close(b.release) }

// TestFullChainDBFenceBlocksHubWritersUntilTickCommits proves, against the real
// committed TiDB-backed store + real Hub + real HubWorkingReader + real
// RecoveryRunner, that every authoritative Hub writer cannot cross the machine
// observation guard write side while RecoveryTick holds the observation lease
// and is blocked before commit. The deterministic hook fires after the row lock;
// the writer must be blocked until the DB blocker is released; the real tick
// then parks the effect and the writer proceeds.
func TestFullChainDBFenceBlocksHubWritersUntilTickCommits(t *testing.T) {
	writers := []struct {
		name   string
		write  func(t *testing.T, hub *Hub, client *clientState, agentID string)
		assert func(t *testing.T, hub *Hub, client *clientState, agentID string)
	}{
		{
			name: "incremental turn update",
			write: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				client.setAgentTurnState(agentID, "active", "sess_fc_1", 8, "sess_fc_1")
			},
			assert: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				state, ok := client.agentTurnState(agentID)
				require.True(t, ok, "the writer's incremental turn update must land after the guard is released")
				require.Equal(t, "active", state)
			},
		},
		{
			name: "ready snapshot running-set replacement",
			write: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				wm := int64(8)
				client.applyReadySnapshot(
					[]string{agentID},
					map[string]struct{}{},
					map[string]string{agentID: "sess_fc_1"},
					map[string]string{agentID: "queued"},
					&wm,
				)
			},
			assert: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				state, ok := client.agentTurnState(agentID)
				require.True(t, ok, "the writer's ready snapshot turn state must land after the guard is released")
				require.Equal(t, "queued", state)
				require.True(t, client.runningAgents[agentID], "the ready snapshot running set must land")
			},
		},
		{
			name: "transport session-generation change",
			write: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				client.setDaemonTransportReady(daemonTransportReady{
					SchemaVersion:     2,
					SessionGeneration: "sess_fc_2",
				})
			},
			assert: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				require.Equal(t, "sess_fc_2", client.currentSessionGeneration(), "the transport session-generation change must land")
			},
		},
		{
			name: "reconnect register replacement",
			write: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				replacement := &clientState{
					machineID:             client.machineID,
					workspaceID:           client.workspaceID,
					conn:                  newProtocolTestWebSocketPair(t),
					sessionGeneration:     "sess_fc_3",
					runningAgents:         map[string]bool{},
					startingAgents:        map[string]bool{},
					stoppedAgents:         map[string]bool{},
					agentStatuses:         map[string]string{},
					agentActivities:       map[string]agentActivityState{},
					agentSessions:         map[string]string{},
					agentTurnStateRecords: map[string]agentTurnStateRecord{},
					connectedAt:           time.Now().UTC(),
				}
				replacement.guard = hub.observationGuard(client.machineID)
				hub.register(replacement)
			},
			assert: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				require.Equal(t, "sess_fc_3", hub.client(client.machineID).currentSessionGeneration(), "the reconnect replacement must land")
			},
		},
		{
			name: "disconnect",
			write: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				require.True(t, hub.unregister(client), "disconnect must remove the current connection")
			},
			assert: func(t *testing.T, hub *Hub, client *clientState, agentID string) {
				require.Nil(t, hub.client(client.machineID), "the disconnect must land after the guard is released")
			},
		},
	}

	for _, writer := range writers {
		t.Run(writer.name, func(t *testing.T) {
			ctx := context.Background()
			pool := openDaemonTestTiDB(t)
			fixture := seedFullChainDBRun(t, pool, writerSuffix(writer.name))

			barrier := newDBBarrier()
			store := run.NewWithOptions(pool, run.StoreOptions{
				WorkspaceGroupID:        fixture.GroupID,
				TestRecoveryTickBarrier: barrier.hook,
			})
			// Arm the park FSM through the production API so the idle observation
			// confirms parked.
			armed, err := store.ParkAwaitingCallbackRecovery(ctx, fixture.WorkspaceID, fixture.RunID, fixture.EffectID, fixture.Now)
			require.NoError(t, err)
			require.True(t, armed, "the awaiting callback must arm the park FSM")

			hub, client := newFullChainDBHub(t, fixture, true)
			reader := agentruntime.NewHubWorkingReader(fullChainDBResolver{fixture: fixture}, hub)
			writerRec := &fullChainDBWriter{}
			runner, err := orchestrator.NewRecoveryRunner(store, writerRec, orchestrator.RecoveryRunnerOptions{
				WorkerID:       "recovery-full-chain-db",
				RuntimeWorking: reader,
			})
			require.NoError(t, err)

			runDone := make(chan error, 1)
			go func() {
				_, err := runner.RecoverOnce(ctx)
				runDone <- err
			}()

			// Wait for the tick to lock the effect/Run rows and block before commit.
			select {
			case <-barrier.entered:
			case <-time.After(10 * time.Second):
				t.Fatal("RecoveryTick never reached the row-locked barrier")
			}

			// Start the authoritative writer: it must block on the guard write side.
			// Instrument the guard's writeWaiting stage so the test can prove the
			// writer reached the write boundary deterministically (no fixed sleep).
			writerStarted := make(chan struct{})
			writerWaiting := make(chan struct{})
			writerDone := make(chan struct{})
			if client.guard != nil {
				client.guard.writeWaiting = func() { close(writerWaiting) }
			}
			go func() {
				close(writerStarted)
				writer.write(t, hub, client, fixture.AgentID)
				close(writerDone)
			}()
			<-writerStarted

			// The writer has reached the machine observation guard write boundary and
			// is blocked there. Non-blockingly prove it has not acquired/completed
			// while the tick holds the observation lease.
			select {
			case <-writerWaiting:
			case <-time.After(10 * time.Second):
				t.Fatal("the writer never reached the machine observation guard write boundary")
			}
			select {
			case <-writerDone:
				t.Fatal("authoritative writer crossed the observation guard while RecoveryTick was blocked before commit")
			default:
			}

			// Release the DB blocker: the real tick commits (parks the effect),
			// the runner releases the observation lease, and the writer proceeds.
			barrier.unblock()
			select {
			case err := <-runDone:
				require.NoError(t, err, "RecoverOnce must succeed after the barrier releases")
			case <-time.After(10 * time.Second):
				t.Fatal("RecoverOnce never returned after the barrier released")
			}
			select {
			case <-writerDone:
			case <-time.After(10 * time.Second):
				t.Fatal("the blocked writer never proceeded after the observation lease was released")
			}

			// The exact durable park state: the idle observation parked the effect.
			effect, err := store.LoadEffect(ctx, fixture.WorkspaceID, fixture.RunID, fixture.EffectID)
			require.NoError(t, err)
			require.Equal(t, run.EffectParkStateParked, effect.ParkState, "the idle observation must confirm parked")
			require.NotNil(t, effect.ParkParkedAt, "park_parked_at must be set")
			require.Equal(t, fixture.AgentID, effect.CallbackResponderAgentID)

			// The writer's authoritative change landed after the fence.
			writer.assert(t, hub, client, fixture.AgentID)
		})
	}
}

// TestFullChainDBFenceDBErrorReleasesLease proves the DB-error path: the barrier
// returns an error after the row lock, RecoveryTick rolls back, the runner
// releases the observation lease, and a blocked writer proceeds — with the
// durable park state unchanged (rollback, zero transition).
func TestFullChainDBFenceDBErrorReleasesLease(t *testing.T) {
	ctx := context.Background()
	pool := openDaemonTestTiDB(t)
	fixture := seedFullChainDBRun(t, pool, "dberr")

	barrier := newDBBarrier()
	barrier.err = fmt.Errorf("forced DB failure at the row-locked boundary")
	store := run.NewWithOptions(pool, run.StoreOptions{
		WorkspaceGroupID:        fixture.GroupID,
		TestRecoveryTickBarrier: barrier.hook,
	})
	armed, err := store.ParkAwaitingCallbackRecovery(ctx, fixture.WorkspaceID, fixture.RunID, fixture.EffectID, fixture.Now)
	require.NoError(t, err)
	require.True(t, armed)

	hub, client := newFullChainDBHub(t, fixture, true)
	reader := agentruntime.NewHubWorkingReader(fullChainDBResolver{fixture: fixture}, hub)
	writerRec := &fullChainDBWriter{}
	runner, err := orchestrator.NewRecoveryRunner(store, writerRec, orchestrator.RecoveryRunnerOptions{
		WorkerID:       "recovery-full-chain-db-error",
		RuntimeWorking: reader,
	})
	require.NoError(t, err)

	runDone := make(chan error, 1)
	go func() {
		_, err := runner.RecoverOnce(ctx)
		runDone <- err
	}()

	select {
	case <-barrier.entered:
	case <-time.After(10 * time.Second):
		t.Fatal("RecoveryTick never reached the row-locked barrier")
	}

	writerWaiting := make(chan struct{})
	writerDone := make(chan struct{})
	if client.guard != nil {
		client.guard.writeWaiting = func() { close(writerWaiting) }
	}
	go func() {
		client.setAgentTurnState("agt_fc_dberr", "active", "sess_fc_1", 8, "sess_fc_1")
		close(writerDone)
	}()
	select {
	case <-writerWaiting:
	case <-time.After(10 * time.Second):
		t.Fatal("the writer never reached the machine observation guard write boundary")
	}
	select {
	case <-writerDone:
		t.Fatal("writer crossed the guard while the tick held the lease")
	default:
	}

	barrier.unblock()
	select {
	case err := <-runDone:
		require.Error(t, err, "RecoveryOnce must surface the DB error")
	case <-time.After(10 * time.Second):
		t.Fatal("RecoverOnce never returned after the forced DB error")
	}
	select {
	case <-writerDone:
	case <-time.After(10 * time.Second):
		t.Fatal("the blocked writer never proceeded after the lease release")
	}

	// Rollback: the durable park FSM is unchanged (still armed, never parked).
	effect, err := store.LoadEffect(ctx, fixture.WorkspaceID, fixture.RunID, fixture.EffectID)
	require.NoError(t, err)
	require.Equal(t, run.EffectParkStateArmOnIdle, effect.ParkState, "the DB error must roll back the park transition")
	require.Nil(t, effect.ParkParkedAt, "a rolled-back tick must not write park_parked_at")
}

// TestFullChainDBFenceContextCancellationReleasesLease proves the context
// cancellation path: RecoveryTick aborts, the runner releases the observation
// lease, and the blocked writer proceeds.
func TestFullChainDBFenceContextCancellationReleasesLease(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	pool := openDaemonTestTiDB(t)
	fixture := seedFullChainDBRun(t, pool, "ctxcancel")

	barrier := newDBBarrier()
	store := run.NewWithOptions(pool, run.StoreOptions{
		WorkspaceGroupID:        fixture.GroupID,
		TestRecoveryTickBarrier: barrier.hook,
	})
	armed, err := store.ParkAwaitingCallbackRecovery(ctx, fixture.WorkspaceID, fixture.RunID, fixture.EffectID, fixture.Now)
	require.NoError(t, err)
	require.True(t, armed)

	hub, client := newFullChainDBHub(t, fixture, true)
	reader := agentruntime.NewHubWorkingReader(fullChainDBResolver{fixture: fixture}, hub)
	writerRec := &fullChainDBWriter{}
	runner, err := orchestrator.NewRecoveryRunner(store, writerRec, orchestrator.RecoveryRunnerOptions{
		WorkerID:       "recovery-full-chain-db-cancel",
		RuntimeWorking: reader,
	})
	require.NoError(t, err)

	runDone := make(chan error, 1)
	go func() {
		_, err := runner.RecoverOnce(ctx)
		runDone <- err
	}()

	select {
	case <-barrier.entered:
	case <-time.After(10 * time.Second):
		t.Fatal("RecoveryTick never reached the row-locked barrier")
	}

	writerWaiting := make(chan struct{})
	writerDone := make(chan struct{})
	if client.guard != nil {
		client.guard.writeWaiting = func() { close(writerWaiting) }
	}
	go func() {
		client.setAgentTurnState("agt_fc_cancel", "active", "sess_fc_1", 8, "sess_fc_1")
		close(writerDone)
	}()
	select {
	case <-writerWaiting:
	case <-time.After(10 * time.Second):
		t.Fatal("the writer never reached the machine observation guard write boundary")
	}
	select {
	case <-writerDone:
		t.Fatal("writer crossed the guard while the tick held the lease")
	default:
	}

	// Cancel the runner context: the tick aborts and the lease releases.
	cancel()
	select {
	case <-runDone:
	case <-time.After(10 * time.Second):
		t.Fatal("RecoverOnce never returned after context cancellation")
	}
	select {
	case <-writerDone:
	case <-time.After(10 * time.Second):
		t.Fatal("the blocked writer never proceeded after the lease release")
	}

	// Cancellation is a rollback: the durable park FSM stays armed.
	effect, err := store.LoadEffect(context.Background(), fixture.WorkspaceID, fixture.RunID, fixture.EffectID)
	require.NoError(t, err)
	require.Equal(t, run.EffectParkStateArmOnIdle, effect.ParkState, "context cancellation must roll back the park transition")
	require.Nil(t, effect.ParkParkedAt)
}

func writerSuffix(name string) string {
	runes := []rune(name)
	if len(runes) == 0 {
		return "x"
	}
	out := make([]rune, 0, 8)
	for _, r := range runes {
		switch {
		case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
			out = append(out, r)
		default:
			out = append(out, '_')
		}
		if len(out) == 8 {
			break
		}
	}
	return string(out)
}