fix: reconcile interrupted shutdown delivery without replay

apps/backend/internal/agent/runtime/lifecycle/already_resumed_recovery_linux_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"syscall"
 	"testing"
 )
@@ -20,7 +21,7 @@ type loadedNativeDeliveryReconciler interface {
 }
 
 func TestRestartRecoveryRetiresOnlyProvenDeadLoadedConversation(t *testing.T) {
-	for _, condition := range []string{"dead owner", "live owner", "missing proof", "different native", "active replacement"} {
+	for _, condition := range []string{"dead owner", "live owner", "missing proof", "different native", "active replacement", "legacy dead owner", "legacy live owner", "legacy missing proof"} {
 		t.Run(condition, func(t *testing.T) {
 			ctx := context.Background()
 			child := exec.Command("/bin/sleep", "60")
@@ -29,7 +30,7 @@ func TestRestartRecoveryRetiresOnlyProvenDeadLoadedConversation(t *testing.T) {
 			t.Cleanup(func() { _ = child.Process.Kill(); _ = child.Wait() })
 			birth, err := processidentity.Capture(child.Process.Pid)
 			require.NoError(t, err)
-			if condition != "live owner" {
+			if !strings.HasSuffix(condition, "live owner") {
 				require.NoError(t, child.Process.Kill())
 				require.Error(t, child.Wait())
 			}
@@ -38,6 +39,9 @@ func TestRestartRecoveryRetiresOnlyProvenDeadLoadedConversation(t *testing.T) {
 			t.Cleanup(func() { _ = store.Close() })
 			peer := journal.Submission{ID: "interrupted", SessionID: "session", IncarnationID: "incarnation",
 				HarnessGeneration: 2, StreamID: "stream", Hash: "hash", State: journal.SubmissionInterruptedUnknown}
+			if strings.HasPrefix(condition, "legacy") {
+				peer.State = journal.SubmissionFailed
+			}
 			_, err = store.PutSubmission(ctx, peer)
 			require.NoError(t, err)
 			manager := newTestManager(t)
@@ -58,13 +62,25 @@ func TestRestartRecoveryRetiresOnlyProvenDeadLoadedConversation(t *testing.T) {
 			identity := AgentDeliveryRecoveryIdentity{TaskID: "task", SessionID: "session", ExecutionID: "old",
 				IncarnationID: "incarnation", HarnessGeneration: 2, StreamID: "stream", SubmissionID: peer.ID,
 				PromptGeneration: 3, OriginalRuntime: birth}
-			if condition == "missing proof" {
+			if strings.HasSuffix(condition, "missing proof") {
 				identity.OriginalRuntime = processidentity.Identity{}
 			}
 			reconciler, supported := any(manager).(loadedNativeDeliveryReconciler)
 			require.True(t, supported, "restart recovery must finish an already loaded native successor")
-			restored, recoverErr := reconciler.ReconcileAlreadyResumedDelivery(ctx, identity, "native")
-			if condition == "dead owner" {
+			restored, recoverErr := false, error(nil)
+			if strings.HasPrefix(condition, "legacy") {
+				manager.runningWriter = &captureExecutorRunningWriter{prior: &models.ExecutorRunning{TaskID: "task", SessionID: "session", AgentExecutionID: "replacement", Metadata: map[string]any{deliveryPromptTurnBindingMetadataKey: deliveryPromptTurnBinding{
+					TaskID: "task", SessionID: "session", ExecutionID: "old", IncarnationID: "incarnation", HarnessGeneration: 2, StreamID: "stream", SubmissionID: peer.ID, TurnID: "turn", PromptGeneration: 3, OriginalRuntime: identity.OriginalRuntime}}}}
+				automatic, ok := any(manager).(interface {
+					ReconcileLoadedNativeDelivery(context.Context, string) (*models.NativeResumeDeliveryEvidence, error)
+				})
+				require.True(t, ok, "loaded replacement must reconcile without recovery presentation metadata")
+				proof, err := automatic.ReconcileLoadedNativeDelivery(ctx, "session")
+				restored, recoverErr = proof != nil, err
+			} else {
+				restored, recoverErr = reconciler.ReconcileAlreadyResumedDelivery(ctx, identity, "native")
+			}
+			if strings.HasSuffix(condition, "dead owner") {
 				require.NoError(t, recoverErr)
 				require.True(t, restored)
 			} else {
@@ -73,8 +89,8 @@ func TestRestartRecoveryRetiresOnlyProvenDeadLoadedConversation(t *testing.T) {
 			}
 			saved, err := store.GetSubmission(ctx, peer.ID)
 			require.NoError(t, err)
-			require.Equal(t, condition == "dead owner", saved.Retired)
-			require.Equal(t, journal.SubmissionInterruptedUnknown, saved.State)
+			require.Equal(t, strings.HasSuffix(condition, "dead owner"), saved.Retired)
+			require.Equal(t, peer.State, saved.State)
 			current, exists := manager.GetExecutionBySessionID("session")
 			require.True(t, exists)
 			require.Same(t, replacement, current)

apps/backend/internal/agent/runtime/lifecycle/delivery_record_evidence.go

@@ -27,6 +27,7 @@ var (
 // historical execution whose process proof may be available. It contains no
 // storage path and is not a browser-facing request.
 type DeliveryRecordEvidenceRequest struct {
+	StoredSubmissionID  string
 	RetiredSubmissionID string
 	TaskID              string
 	SessionID           string
@@ -39,6 +40,7 @@ type DeliveryRecordEvidenceRequest struct {
 // DeliveryRecordEvidence separates the live session instance from proof about
 // the historical process that owned a candidate prompt.
 type DeliveryRecordEvidence struct {
+	StoredSubmission  *journal.Submission
 	RetiredSubmission *journal.Submission
 	Descriptor        journal.RecoveryDescriptor
 	ProcessTerminated *bool
@@ -93,6 +95,9 @@ func (m *Manager) InspectDeliveryRecordEvidence(
 		return result, errors.Join(ErrDeliveryRecordEvidenceTruncated, ErrDeliveryRecoveryBlocked)
 	}
 	result.ControlIdentity = m.deliveryRecordControlIdentity(ctx, request, result.Descriptor)
+	if result.ControlIdentity == nil {
+		result.ControlIdentity = m.targetedDeliveryRecordControlIdentity(ctx, request, result)
+	}
 	if identity := result.ControlIdentity; identity != nil {
 		result.ProcessTerminated = historicalProcessTermination(DeliveryRecordEvidenceRequest{
 			ExecutionID: identity.ExecutionID, OriginalRuntime: &identity.OriginalRuntime,
@@ -271,3 +276,14 @@ func historicalProcessTermination(request DeliveryRecordEvidenceRequest, liveExe
 	}
 	return &terminated
 }
+
+func (m *Manager) targetedDeliveryRecordControlIdentity(ctx context.Context, request DeliveryRecordEvidenceRequest, result *DeliveryRecordEvidence) *AgentDeliveryRecoveryIdentity {
+	if result.Descriptor.SubmissionCount == 0 && result.StoredSubmission != nil {
+		selected := result.StoredSubmission
+		if selected.ID == request.StoredSubmissionID && selected.SessionID == request.SessionID && selected.IncarnationID == request.IncarnationID && selected.HarnessGeneration == request.HarnessGeneration {
+			return m.persistedDeliveryRecordControlIdentity(ctx, request, journal.SubmissionSummary{
+				ID: selected.ID, SessionID: selected.SessionID, IncarnationID: selected.IncarnationID, HarnessGeneration: selected.HarnessGeneration, StreamID: selected.StreamID})
+		}
+	}
+	return nil
+}

apps/backend/internal/agent/runtime/lifecycle/delivery_record_transport.go

@@ -16,13 +16,13 @@ func (m *Manager) inspectDeliveryRecordTransport(ctx context.Context, lease *age
 func (m *Manager) inspectRetainedDeliveryRecord(ctx context.Context, lease *agentctl.RuntimeLease, control *agentctl.ControlClient, request DeliveryRecordEvidenceRequest) (*DeliveryRecordEvidence, error) {
 	retained, readErr := control.ReadRetainedReconstructionEvidence(ctx, journal.RetainedReconstructionEvidenceRequest{
 		Root: m.dataDir, SessionID: request.SessionID, IncarnationID: request.IncarnationID,
-		HarnessGeneration: request.HarnessGeneration, OriginalRuntime: request.OriginalRuntime, RetiredSubmissionID: request.RetiredSubmissionID,
+		HarnessGeneration: request.HarnessGeneration, OriginalRuntime: request.OriginalRuntime, RetiredSubmissionID: request.RetiredSubmissionID, StoredSubmissionID: request.StoredSubmissionID,
 	})
 	if readErr != nil {
 		return nil, deliveryEvidenceReadError(lease, readErr)
 	}
 	return &DeliveryRecordEvidence{
-		Descriptor: retained.Descriptor, ProcessTerminated: retained.ProcessTerminated, RetiredSubmission: retained.RetiredSubmission,
+		Descriptor: retained.Descriptor, ProcessTerminated: retained.ProcessTerminated, RetiredSubmission: retained.RetiredSubmission, StoredSubmission: retained.StoredSubmission,
 	}, nil
 }
 
@@ -44,20 +44,8 @@ func (m *Manager) inspectLiveDeliveryRecord(ctx context.Context, lease *agentctl
 		Descriptor: *descriptor, LiveExecutionID: instance.ID,
 		ProcessTerminated: historicalProcessTermination(request, instance.ID),
 	}
-	if request.RetiredSubmissionID != "" {
-		retired, err := client.GetDeliverySubmission(ctx, request.RetiredSubmissionID)
-		if err != nil {
-			return nil, deliveryEvidenceReadError(lease, err)
-		}
-		after, err := client.GetDeliveryStatus(ctx, "")
-		if err != nil {
-			return nil, deliveryEvidenceReadError(lease, err)
-		}
-		if !sameDeliveryRecordDescriptor(*descriptor, *after) {
-			return nil, ErrDeliveryRecordEvidenceStale
-		}
-		retired.Payload = nil
-		result.RetiredSubmission = retired
+	if err := m.readLiveStoredEvidence(ctx, lease, client, request, result); err != nil {
+		return nil, err
 	}
 	return result, nil
 }
@@ -147,3 +135,47 @@ func (m *Manager) readLiveDeliveryRecord(ctx context.Context, lease *agentctl.Ru
 	after = *second
 	return &deliveryRecordRead{before: before, after: after, submission: submission}, nil
 }
+
+func (m *Manager) readLiveStoredEvidence(ctx context.Context, lease *agentctl.RuntimeLease, client *agentctl.Client, request DeliveryRecordEvidenceRequest, result *DeliveryRecordEvidence) error {
+	for _, id := range []string{request.RetiredSubmissionID, request.StoredSubmissionID} {
+		if id == "" {
+			continue
+		}
+		record, err := readLiveSubmissionMetadata(ctx, client, request, id)
+		if err != nil {
+			return deliveryEvidenceReadError(lease, err)
+		}
+		if id == request.RetiredSubmissionID {
+			if !record.Retired || !journal.HasUnprovenOutcome(*record) {
+				return ErrDeliveryOwnerMismatch
+			}
+			result.RetiredSubmission = record
+		}
+		if id == request.StoredSubmissionID {
+			result.StoredSubmission = record
+		}
+	}
+	if request.RetiredSubmissionID == "" && request.StoredSubmissionID == "" {
+		return nil
+	}
+	after, err := client.GetDeliveryStatus(ctx, "")
+	if err != nil {
+		return deliveryEvidenceReadError(lease, err)
+	}
+	if !sameNativeResumeProofStatus(&result.Descriptor, after) {
+		return ErrDeliveryRecordEvidenceStale
+	}
+	return nil
+}
+
+func readLiveSubmissionMetadata(ctx context.Context, client *agentctl.Client, request DeliveryRecordEvidenceRequest, id string) (*journal.Submission, error) {
+	record, err := client.GetDeliverySubmission(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	if record == nil || record.SessionID != request.SessionID || record.IncarnationID != request.IncarnationID || record.HarnessGeneration == 0 || record.HarnessGeneration > request.HarnessGeneration {
+		return nil, ErrDeliveryOwnerMismatch
+	}
+	record.Payload = nil
+	return record, nil
+}

apps/backend/internal/agent/runtime/lifecycle/e2e_pruned_delivery_fixture_test.go

@@ -93,8 +93,7 @@ func assertPrunedDeliveryRestored(t *testing.T, repo *tasksqlite.Repository, ses
 	original, err := repo.GetAgentDeliverySubmission(ctx, "prompt:message-e2e-record-"+sessionID+"-1")
 	require.NoError(t, err)
 	require.Equal(t, models.DeliverySubmissionInterruptedUnknown, original.State)
-	// Reconstruction preserves the existing outcome; native-resume reconciliation records unknown.
-	require.Contains(t, []string{"", "unknown"}, original.Outcome)
+	require.Equal(t, "unknown", original.Outcome)
 	require.Zero(t, original.TerminalSequence)
 	generation, err := repo.GetCurrentHarnessSessionGeneration(ctx, sessionID, original.IncarnationID)
 	require.NoError(t, err)

apps/backend/internal/agent/runtime/lifecycle/loaded_native_reconciliation.go

@@ -0,0 +1,95 @@
+package lifecycle
+
+import (
+	"context"
+	"fmt"
+
+	agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
+	"github.com/kandev/kandev/internal/agentctl/journal"
+	"github.com/kandev/kandev/internal/task/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+// ReconcileLoadedNativeDelivery completes retirement for an idle native successor.
+// The persisted prompt binding proves the original owner, independently of the
+// replacement's current process and of missing UI recovery metadata.
+func (m *Manager) ReconcileLoadedNativeDelivery(ctx context.Context, sessionID string) (*models.NativeResumeDeliveryEvidence, error) {
+	execution, exists := m.GetExecutionBySessionID(sessionID)
+	if !exists || execution == nil {
+		return nil, ErrDurableAdoptionEvidenceUnavailable
+	}
+	if !execution.promptMu.TryLock() {
+		return nil, ErrDeliveryRecoveryBlocked
+	}
+	defer execution.promptMu.Unlock()
+	if execution.Status != v1.AgentStatusReady || execution.dispatchedPromptPending.Load() {
+		return nil, ErrDeliveryRecoveryBlocked
+	}
+	if proof, err := m.GetNativeResumeDeliveryEvidence(ctx, sessionID); err == nil && proof != nil {
+		return proof, nil
+	}
+	historical, err := m.readLoadedNativeHistoricalOwner(ctx, execution)
+	if err != nil {
+		return nil, err
+	}
+	if historical == nil || historical.ExecutionID == execution.ID || !loadedRecoverySuccessorMatches(execution, *historical) {
+		return nil, ErrDeliveryRecoveryBlocked
+	}
+	if err := proveHistoricalDeliveryOwnerDead(*historical); err != nil {
+		return nil, err
+	}
+	if err := m.AcknowledgeNativeResumeDelivery(ctx, sessionID); err != nil {
+		return nil, err
+	}
+	return m.GetNativeResumeDeliveryEvidence(ctx, sessionID)
+}
+
+func (m *Manager) readLoadedNativeHistoricalOwner(ctx context.Context, execution *AgentExecution) (*AgentDeliveryRecoveryIdentity, error) {
+	client, release := execution.AcquireAgentCtlClient()
+	defer release()
+	if client == nil {
+		return nil, ErrDurableAdoptionEvidenceUnavailable
+	}
+	status, err := m.nativeResumeDeliveryStatus(ctx, execution, client)
+	if err != nil || status == nil {
+		return nil, err
+	}
+	if m.streamManager == nil || m.streamManager.deliveryRepository() == nil {
+		return nil, ErrDurableAdoptionEvidenceUnavailable
+	}
+	peer, err := client.ListDeliverySubmissions(ctx, execution.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	rows, err := activeBackendSubmissions(ctx, m.streamManager.deliveryRepository(), execution.SessionID)
+	if err != nil {
+		return nil, err
+	}
+	return m.loadedNativeHistoricalOwner(ctx, execution, status, peer, rows)
+}
+func (m *Manager) loadedNativeHistoricalOwner(ctx context.Context, execution *AgentExecution, status *agentctl.DeliveryStatus, peer []journal.Submission, rows []*models.AgentDeliverySubmission) (*AgentDeliveryRecoveryIdentity, error) {
+	var historical *AgentDeliveryRecoveryIdentity
+	for _, row := range rows {
+		for _, submission := range peer {
+			if submission.ID != row.ID || submission.Retired {
+				continue
+			}
+			if historical != nil {
+				return nil, fmt.Errorf("%w: independent unretired history", ErrDeliveryRecoveryBlocked)
+			}
+			candidate := *row
+			if candidate.State == models.DeliverySubmissionDispatching && submission.State == journal.SubmissionInterruptedUnknown {
+				candidate.State = models.DeliverySubmissionInterruptedUnknown
+			}
+			if !peerMatchesUnknownHistory(execution, status, submission, &candidate) {
+				return nil, ErrDeliveryOwnerMismatch
+			}
+			historical = m.persistedDeliveryRecordControlIdentity(ctx, DeliveryRecordEvidenceRequest{TaskID: execution.TaskID, SessionID: execution.SessionID}, journal.SubmissionSummary{
+				ID: submission.ID, SessionID: submission.SessionID, IncarnationID: submission.IncarnationID, HarnessGeneration: submission.HarnessGeneration, StreamID: submission.StreamID})
+			if historical == nil {
+				return nil, ErrDeliveryRecoveryBlocked
+			}
+		}
+	}
+	return historical, nil
+}

apps/backend/internal/agent/runtime/lifecycle/native_resume_adoption.go

@@ -64,9 +64,10 @@ func peerAcknowledgesUnknownHistory(execution *AgentExecution, status *agentctl.
 
 func peerMatchesUnknownHistory(execution *AgentExecution, status *agentctl.DeliveryStatus, peer journal.Submission, submission *models.AgentDeliverySubmission) bool {
 	return submission.State == models.DeliverySubmissionInterruptedUnknown &&
-		peer.ID == submission.ID && peer.State == journal.SubmissionInterruptedUnknown &&
+		(peer.State == journal.SubmissionInterruptedUnknown || submission.TerminalSequence == 0) && peer.ID == submission.ID && journal.HasUnprovenOutcome(peer) &&
 		peer.SessionID == execution.SessionID && peer.SessionID == submission.SessionID &&
 		peer.IncarnationID == status.IncarnationID && peer.IncarnationID == submission.IncarnationID &&
 		peer.HarnessGeneration > 0 && peer.HarnessGeneration <= status.HarnessGeneration && int64(peer.HarnessGeneration) == submission.HarnessGeneration &&
-		peer.Hash != "" && peer.Hash == submission.PayloadHash
+		peer.Hash != "" && peer.Hash == submission.PayloadHash &&
+		(peer.HarnessGeneration != status.HarnessGeneration || peer.StreamID == status.StreamID)
 }

apps/backend/internal/agent/runtime/lifecycle/native_resume_delivery.go

@@ -8,6 +8,7 @@ import (
 
 	agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
 	"github.com/kandev/kandev/internal/agentctl/journal"
+	"github.com/kandev/kandev/internal/task/models"
 )
 
 // AcknowledgeNativeResumeDelivery seals uncertain prior prompts after an
@@ -36,10 +37,18 @@ func (m *Manager) AcknowledgeNativeResumeDelivery(ctx context.Context, sessionID
 	if err := m.reconcileNativeResumeSQLDispatch(ctx, execution, status, submissions); err != nil {
 		return err
 	}
-	if err := m.validateNativeResumeSQLHistory(ctx, execution, status, submissions, false); err != nil {
+	if _, err := authenticatedDeliveryReplayCursor(ctx, execution, m.streamManager.deliveryRepository(), status); err != nil {
 		return err
 	}
-	ids, err := nativeResumeInterruptedSubmissionIDs(execution, submissions)
+	// Dispatch reconciliation can change SQL state; pin one fresh inventory for validation and retirement selection.
+	rows, err := activeBackendSubmissions(ctx, m.streamManager.deliveryRepository(), sessionID)
+	if err != nil {
+		return err
+	}
+	if err := validateNativeResumeSQLRows(execution, status, submissions, rows, false); err != nil {
+		return err
+	}
+	ids, err := nativeResumeRetirementIDs(execution, status, submissions, rows)
 	if err != nil {
 		return err
 	}
@@ -79,25 +88,36 @@ func validateNativeResumeDeliveryOwner(execution *AgentExecution, status *agentc
 	return nil
 }
 
-func nativeResumeInterruptedSubmissionIDs(execution *AgentExecution, submissions []journal.Submission) ([]string, error) {
+func nativeResumeRetirementIDs(execution *AgentExecution, status *agentctl.DeliveryStatus, submissions []journal.Submission, rows []*models.AgentDeliverySubmission) ([]string, error) {
 	var ids []string
+	backend := make(map[string]*models.AgentDeliverySubmission, len(rows))
+	for _, row := range rows {
+		backend[row.ID] = row
+	}
 	for _, submission := range submissions {
+		if submission.Retired {
+			continue
+		}
 		switch submission.State {
 		case journal.SubmissionPrepared, journal.SubmissionAccepted, journal.SubmissionDispatching:
 			return nil, fmt.Errorf("native resume cannot acknowledge live delivery work")
-		case journal.SubmissionCompleted:
-			if !submission.TerminalEventRetained {
-				return nil, fmt.Errorf("native resume cannot acknowledge incomplete terminal evidence")
-			}
-		case journal.SubmissionInterruptedUnknown:
-			if submission.Retired {
+		case journal.SubmissionInterruptedUnknown, journal.SubmissionCompleted, journal.SubmissionFailed, journal.SubmissionCancelled:
+			if !journal.HasUnprovenOutcome(submission) {
 				continue
 			}
-			if submission.SessionID != execution.SessionID || submission.IncarnationID != execution.DeliveryIncarnationID ||
-				submission.HarnessGeneration == 0 || submission.HarnessGeneration > execution.DeliveryHarnessGeneration {
-				return nil, journal.ErrOwnerMismatch
+			sqlSubmission, found := backend[submission.ID]
+			if !found || sqlSubmission.State != models.DeliverySubmissionInterruptedUnknown {
+				if submission.State == journal.SubmissionFailed || submission.State == journal.SubmissionCancelled {
+					continue
+				}
+				return nil, fmt.Errorf("%w: native resume requires uncertain SQL history", ErrDurableAdoptionBlocked)
+			}
+			if !peerMatchesUnknownHistory(execution, status, submission, sqlSubmission) {
+				return nil, fmt.Errorf("%w: native resume submission identity changed", ErrDurableAdoptionBlocked)
 			}
 			ids = append(ids, submission.ID)
+		default:
+			return nil, fmt.Errorf("%w: unsupported native resume submission state", ErrDurableAdoptionBlocked)
 		}
 	}
 	return ids, nil