feat(localtaskstore): add durable local state for task receipt, outbox, and restart recovery

app/services/localtaskstore/execution_test.go

@@ -0,0 +1,68 @@
+package localtaskstore
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRecordReceivedSurvivesRestart(t *testing.T) {
+	storePath := filepath.Join(t.TempDir(), "task_store.db")
+	store := openTestStore(t, storePath, 1024*1024, 1024)
+
+	received, err := store.RecordReceived(TaskReceipt{
+		TaskID:             "task-1",
+		ExecutionAttemptID: "attempt-1",
+	})
+	require.NoError(t, err)
+	require.Equal(t, TaskStatusReceived, received.Status)
+	require.NoError(t, store.Close())
+
+	reopened := openTestStore(t, storePath, 1024*1024, 1024)
+	state, err := reopened.TaskState("task-1", "attempt-1")
+	require.NoError(t, err)
+	require.True(t, state.Exists)
+	require.Equal(t, TaskStatusReceived, state.Status)
+}
+
+func TestRecordReceivedReturnsExistingDuplicateState(t *testing.T) {
+	store := newTestStore(t, 1024*1024, 1024)
+	receipt := TaskReceipt{TaskID: "task-1", ExecutionAttemptID: "attempt-1"}
+
+	first, err := store.RecordReceived(receipt)
+	require.NoError(t, err)
+	second, err := store.RecordReceived(receipt)
+	require.NoError(t, err)
+
+	require.True(t, second.Exists)
+	require.Equal(t, first.ID, second.ID)
+	require.Equal(t, TaskStatusReceived, second.Status)
+}
+
+func TestTaskStateTreatsNewAttemptAsDistinct(t *testing.T) {
+	store := newTestStore(t, 1024*1024, 1024)
+
+	_, err := store.RecordReceived(TaskReceipt{TaskID: "task-1", ExecutionAttemptID: "attempt-1"})
+	require.NoError(t, err)
+
+	state, err := store.TaskState("task-1", "attempt-2")
+	require.NoError(t, err)
+	require.False(t, state.Exists)
+}
+
+func openTestStore(t *testing.T, path string, spoolCapBytes, terminalReserveBytes int64) *Store {
+	t.Helper()
+
+	store, err := New(Config{
+		Path:                 path,
+		SpoolCapBytes:        spoolCapBytes,
+		TerminalReserveBytes: terminalReserveBytes,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = store.Close()
+	})
+
+	return store
+}

app/services/localtaskstore/outbox_test.go

@@ -0,0 +1,87 @@
+package localtaskstore
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRecordStartedSurvivesRestart(t *testing.T) {
+	storePath := filepath.Join(t.TempDir(), "task_store.db")
+	store := openTestStore(t, storePath, 1024*1024, 1024)
+
+	_, err := store.RecordReceived(TaskReceipt{TaskID: "task-1", ExecutionAttemptID: "attempt-1"})
+	require.NoError(t, err)
+	require.NoError(t, store.RecordStarted("task-1", "attempt-1"))
+	require.NoError(t, store.Close())
+
+	reopened := openTestStore(t, storePath, 1024*1024, 1024)
+	state, err := reopened.TaskState("task-1", "attempt-1")
+	require.NoError(t, err)
+	require.Equal(t, TaskStatusRunning, state.Status)
+}
+
+func TestRecordFinalCreatesUnackedTerminalMessageAcrossRestart(t *testing.T) {
+	storePath := filepath.Join(t.TempDir(), "task_store.db")
+	store := openTestStore(t, storePath, 1024*1024, 1024)
+
+	require.NoError(t, store.RecordFinal(FinalResult{
+		MessageID:          "msg-final-1",
+		TaskID:             "task-1",
+		ExecutionAttemptID: "attempt-1",
+		Status:             "completed",
+		ExitCode:           0,
+		Payload:            `{"status":"completed"}`,
+	}))
+	require.NoError(t, store.Close())
+
+	reopened := openTestStore(t, storePath, 1024*1024, 1024)
+	messages, err := reopened.UnackedMessages()
+	require.NoError(t, err)
+	require.Len(t, messages, 1)
+	require.Equal(t, "msg-final-1", messages[0].MessageID)
+	require.Equal(t, OutboxMessageTypeFinal, messages[0].Type)
+}
+
+func TestAckMessageRemovesOutputChunkFromResendQueue(t *testing.T) {
+	store := newTestStore(t, 1024*1024, 1024)
+
+	require.NoError(t, store.AppendOutputChunk(OutputChunk{
+		MessageID:          "msg-output-1",
+		TaskID:             "task-1",
+		ExecutionAttemptID: "attempt-1",
+		Stream:             "stdout",
+		Sequence:           1,
+		Payload:            "hello",
+		ByteCount:          5,
+	}))
+	require.NoError(t, store.AckMessage("msg-output-1"))
+
+	messages, err := store.UnackedMessages()
+	require.NoError(t, err)
+	require.Empty(t, messages)
+}
+
+func TestAckFinalRemovesOutboxButPreservesTaskState(t *testing.T) {
+	store := newTestStore(t, 1024*1024, 1024)
+
+	require.NoError(t, store.RecordFinal(FinalResult{
+		MessageID:          "msg-final-1",
+		TaskID:             "task-1",
+		ExecutionAttemptID: "attempt-1",
+		Status:             "completed",
+		ExitCode:           0,
+		Payload:            `{"status":"completed"}`,
+	}))
+	require.NoError(t, store.AckMessage("msg-final-1"))
+
+	messages, err := store.UnackedMessages()
+	require.NoError(t, err)
+	require.Empty(t, messages)
+
+	state, err := store.TaskState("task-1", "attempt-1")
+	require.NoError(t, err)
+	require.True(t, state.Exists)
+	require.Equal(t, TaskStatusFinal, state.Status)
+}

app/services/localtaskstore/recovery_test.go

@@ -0,0 +1,54 @@
+package localtaskstore
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSnapshotIncludesLocalTruncationState(t *testing.T) {
+	store := newTestStore(t, 16, 4)
+	appendChunk(t, store, "msg-1", "task-1", 1, "12345")
+	appendChunk(t, store, "msg-2", "task-1", 2, "67890")
+	appendChunk(t, store, "msg-3", "task-1", 3, "abcde")
+
+	snapshot, err := store.Snapshot()
+	require.NoError(t, err)
+	require.Len(t, snapshot.Tasks, 1)
+	require.Equal(t, "task-1", snapshot.Tasks[0].TaskID)
+	require.True(t, snapshot.Tasks[0].LocalOutputTruncated)
+}
+
+func TestMarkInterruptedRunningTasksQueuesTerminalRecordAcrossRestart(t *testing.T) {
+	storePath := filepath.Join(t.TempDir(), "task_store.db")
+	store := openTestStore(t, storePath, 1024*1024, 1024)
+
+	require.NoError(t, store.RecordStarted("task-1", "attempt-1"))
+	require.NoError(t, store.Close())
+
+	reopened := openTestStore(t, storePath, 1024*1024, 1024)
+	require.NoError(t, reopened.MarkInterruptedRunningTasks())
+
+	state, err := reopened.TaskState("task-1", "attempt-1")
+	require.NoError(t, err)
+	require.Equal(t, TaskStatusInterrupted, state.Status)
+
+	messages, err := reopened.UnackedMessages()
+	require.NoError(t, err)
+	require.Len(t, messages, 1)
+	require.Equal(t, OutboxMessageTypeFinal, messages[0].Type)
+	require.Contains(t, messages[0].Payload, "interrupted")
+}
+
+func TestMarkInterruptedRunningTasksIsIdempotent(t *testing.T) {
+	store := newTestStore(t, 1024*1024, 1024)
+
+	require.NoError(t, store.RecordStarted("task-1", "attempt-1"))
+	require.NoError(t, store.MarkInterruptedRunningTasks())
+	require.NoError(t, store.MarkInterruptedRunningTasks())
+
+	messages, err := store.UnackedMessages()
+	require.NoError(t, err)
+	require.Len(t, messages, 1)
+}

app/services/localtaskstore/spool_test.go

@@ -0,0 +1,81 @@
+package localtaskstore
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAppendOutputChunkRotatesOldestChunksAndMarksTruncated(t *testing.T) {
+	store := newTestStore(t, 16, 4)
+
+	appendChunk(t, store, "msg-1", "task-1", 1, "12345")
+	appendChunk(t, store, "msg-2", "task-1", 2, "67890")
+	appendChunk(t, store, "msg-3", "task-1", 3, "abcde")
+
+	messages, err := store.UnackedMessages()
+	require.NoError(t, err)
+	require.Len(t, messages, 2)
+	require.Equal(t, "msg-2", messages[0].MessageID)
+	require.Equal(t, "msg-3", messages[1].MessageID)
+
+	state, err := store.TaskState("task-1", "attempt-1")
+	require.NoError(t, err)
+	require.True(t, state.LocalOutputTruncated)
+}
+
+func TestRecordFinalPreservedUnderChunkCapPressure(t *testing.T) {
+	store := newTestStore(t, 14, 4)
+
+	appendChunk(t, store, "msg-1", "task-1", 1, "12345")
+	appendChunk(t, store, "msg-2", "task-1", 2, "67890")
+	require.NoError(t, store.RecordFinal(FinalResult{
+		MessageID:          "msg-final-1",
+		TaskID:             "task-1",
+		ExecutionAttemptID: "attempt-1",
+		Status:             "completed",
+		Payload:            "done",
+	}))
+
+	messages, err := store.UnackedMessages()
+	require.NoError(t, err)
+	require.Contains(t, messageIDs(messages), "msg-final-1")
+}
+
+func TestRecordReceivedFailsWhenTerminalReserveUnavailable(t *testing.T) {
+	store := newTestStore(t, 8, 4)
+	require.NoError(t, store.RecordFinal(FinalResult{
+		MessageID:          "msg-final-1",
+		TaskID:             "task-1",
+		ExecutionAttemptID: "attempt-1",
+		Status:             "completed",
+		Payload:            "12345",
+	}))
+
+	_, err := store.RecordReceived(TaskReceipt{TaskID: "task-2", ExecutionAttemptID: "attempt-1"})
+	require.Error(t, err)
+	require.True(t, errors.Is(err, ErrTerminalReserveUnavailable))
+}
+
+func messageIDs(messages []OutboxMessage) []string {
+	ids := make([]string, 0, len(messages))
+	for _, message := range messages {
+		ids = append(ids, message.MessageID)
+	}
+	return ids
+}
+
+func appendChunk(t *testing.T, store *Store, messageID, taskID string, sequence int64, payload string) {
+	t.Helper()
+
+	require.NoError(t, store.AppendOutputChunk(OutputChunk{
+		MessageID:          messageID,
+		TaskID:             taskID,
+		ExecutionAttemptID: "attempt-1",
+		Stream:             "stdout",
+		Sequence:           sequence,
+		Payload:            payload,
+		ByteCount:          int64(len(payload)),
+	}))
+}