fix(checkpoint): allow failed checkpoint cleanup

docs/architecture/persistence-checkpoints-and-forks.md

@@ -86,6 +86,14 @@ deleted while its context and checkpoint remain.
 Deletion first hides the checkpoint, then deletes its snapshot tag, then removes
 the row. A checkpoint referenced by a fork cannot be deleted. Substrate deletes the Tag's copied snapshot with the Tag.
 
+Failed checkpoints can also be deleted by their owner. Creation marks a checkpoint
+failed only after snapshot-tag cleanup succeeds or confirms that no tag exists.
+Deletion verifies that absence before removing the row; it never deletes an
+unexpected tag without a retained UID. Backend errors leave the checkpoint in
+`DELETING` for a later delete request to retry. Its history and runtime revision
+references remain retained until deletion completes. Checkpoints still in
+`CREATING` cannot be deleted through this path.
+
 ## Forking
 
 Forks are currently unavailable. Substrate 0.5.0-alpha2 applies one snapshot scope to pause and suspend, and templates use Full so paused Actors keep their process memory. A Full checkpoint restored into a new Actor would resume processes that still hold the source session's IDs, so `ForkSession` rejects it with `FAILED_PRECONDITION`. Forking returns when Substrate's lifecycle v2 lets templates take Data snapshots on suspend again. The rest of this section describes forks of Data checkpoints.

go/core/internal/database/checkpoints.go

@@ -445,7 +445,7 @@ func (c *Client) ListSessionCheckpoints(ctx context.Context, sessionID, userID,
 	return result, nil
 }
 
-// BeginDeleteSessionCheckpoint marks an owned READY checkpoint DELETING and returns
+// BeginDeleteSessionCheckpoint marks an owned READY or FAILED checkpoint DELETING and returns
 // the snapshot and tag identity for external cleanup. Retrying a DELETING checkpoint is
 // allowed. A fork reference, another lifecycle state, or a missing/unowned checkpoint
 // returns ErrNotFound.
@@ -470,7 +470,7 @@ func (c *Client) BeginDeleteSessionCheckpoint(ctx context.Context, id, userID st
 			UPDATE session_checkpoint
 			SET state = 'DELETING', data = $3
 			WHERE session_checkpoint.id = $1 AND session_checkpoint.user_id = $2
-			  AND session_checkpoint.state IN ('READY', 'DELETING')
+			  AND session_checkpoint.state IN ('READY', 'FAILED', 'DELETING')
 			  AND NOT EXISTS (
 			      SELECT 1 FROM session i WHERE i.pinned_checkpoint_id = session_checkpoint.id
 			  )

go/core/internal/database/runtime_revisions_test.go

@@ -82,63 +82,82 @@ func TestRuntimeRevisionCollectionAfterPairRetirement(t *testing.T) {
 }
 
 func TestRuntimeRevisionCollectionPreservesSessionAndCheckpoint(t *testing.T) {
-	client := NewClient(setupTestDB(t), "public")
-	ctx := t.Context()
-	sessionFixture(t, client, ctx, "team-a", "revision", "assistant", "kagent")
-	session, _, err := client.CreateSession(ctx, newSessionRequest(uuid.NewString(), "assistant", "kagent", ""), "session")
-	require.NoError(t, err)
-	_, err = markSessionReady(ctx, client, session.GetId(), "runtime")
-	require.NoError(t, err)
-	require.NoError(t, client.RetireAgentIdentities(ctx, "team-a", "assistant", nil))
+	for _, tc := range []struct {
+		name, tagUID, snapshotURI, failure string
+	}{
+		{name: "ready", tagUID: "tag-uid", snapshotURI: "s3://tags/checkpoint"},
+		{name: "failed", failure: "tag creation failed"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			client := NewClient(setupTestDB(t), "public")
+			ctx := t.Context()
+			sessionFixture(t, client, ctx, "team-a", "revision", "assistant", "kagent")
+			session, _, err := client.CreateSession(ctx, newSessionRequest(uuid.NewString(), "assistant", "kagent", ""), "session")
+			require.NoError(t, err)
+			_, err = markSessionReady(ctx, client, session.GetId(), "runtime")
+			require.NoError(t, err)
+			require.NoError(t, client.RetireAgentIdentities(ctx, "team-a", "assistant", nil))
 
-	assertRetained := func() {
-		t.Helper()
-		revisions, err := client.ListUnreferencedRuntimeRevisions(ctx)
-		require.NoError(t, err)
-		require.Empty(t, revisions)
+			assertRetained := func() {
+				t.Helper()
+				revisions, err := client.ListUnreferencedRuntimeRevisions(ctx)
+				require.NoError(t, err)
+				require.Empty(t, revisions)
 
-		claimed, err := client.BeginRuntimeRevisionDeletion(ctx, "revision")
-		require.NoError(t, err)
-		require.Nil(t, claimed)
-		require.NoError(t, client.DeleteRuntimeRevision(ctx, "revision", "revision-actor-uid"))
-		_, err = client.GetRuntimeRevision(ctx, "revision")
-		require.NoError(t, err)
-	}
-	assertRetained()
-	task := newSessionTask("task", "message")
-	task.ContextID = session.GetContextId()
-	_, err = client.CreateRuntimeTask(ctx, session.GetId(), taskMutationHash("request"), task, "")
-	require.NoError(t, err)
-	task.Status.State = a2a.TaskStateCompleted
-	require.NoError(t, saveRuntimeTask(t, client, session.GetId(), task, task,
-		&SessionTaskSnapshot{Atespace: "team-a", URI: "s3://snapshots/task", ContentScope: "FULL"}))
-	checkpoint, _, err := client.ReserveSessionCheckpoint(ctx, &apiv1alpha1.Checkpoint{
-		Id: uuid.NewString(), SessionId: session.GetId(), HeadTaskId: string(task.ID),
-	}, "alice", "checkpoint")
-	require.NoError(t, err)
-	require.ErrorIs(t, deleteSession(ctx, client, session.GetId()), ErrConflict)
+				claimed, err := client.BeginRuntimeRevisionDeletion(ctx, "revision")
+				require.NoError(t, err)
+				require.Nil(t, claimed)
+				require.NoError(t, client.DeleteRuntimeRevision(ctx, "revision", "revision-actor-uid"))
+				_, err = client.GetRuntimeRevision(ctx, "revision")
+				require.NoError(t, err)
+			}
+			assertRetained()
+			task := newSessionTask("task", "message")
+			task.ContextID = session.GetContextId()
+			_, err = client.CreateRuntimeTask(ctx, session.GetId(), taskMutationHash("request"), task, "")
+			require.NoError(t, err)
+			task.Status.State = a2a.TaskStateCompleted
+			require.NoError(t, saveRuntimeTask(t, client, session.GetId(), task, task,
+				&SessionTaskSnapshot{Atespace: "team-a", URI: "s3://snapshots/task", ContentScope: "FULL"}))
+			checkpoint, _, err := client.ReserveSessionCheckpoint(ctx, &apiv1alpha1.Checkpoint{
+				Id: uuid.NewString(), SessionId: session.GetId(), HeadTaskId: string(task.ID),
+			}, "alice", "checkpoint")
+			require.NoError(t, err)
+			require.ErrorIs(t, deleteSession(ctx, client, session.GetId()), ErrConflict)
 
-	// A checkpoint retains the runtime throughout creation, use, and deletion,
-	// even after the source session and active template pair are gone.
-	assertRetained()
-	_, err = client.FinalizeSessionCheckpoint(ctx, checkpoint.GetId(), "tag-uid", "s3://tags/checkpoint", "")
-	require.NoError(t, err)
-	require.NoError(t, deleteSession(ctx, client, session.GetId()))
-	assertRetained()
-	_, _, err = client.BeginDeleteSessionCheckpoint(ctx, checkpoint.GetId(), "alice")
-	require.NoError(t, err)
-	assertRetained()
-	require.NoError(t, client.DeleteSessionCheckpoint(ctx, checkpoint.GetId(), "alice"))
-	revisions, err := client.ListUnreferencedRuntimeRevisions(ctx)
-	require.NoError(t, err)
-	require.Len(t, revisions, 1)
+			// A checkpoint retains the runtime throughout creation, use, and deletion,
+			// even after the source session and active template pair are gone.
+			assertRetained()
+			_, _, err = client.BeginDeleteSessionCheckpoint(ctx, checkpoint.GetId(), "alice")
+			require.ErrorIs(t, err, ErrNotFound, "an in-flight checkpoint must not be deleted")
+			_, err = client.FinalizeSessionCheckpoint(ctx, checkpoint.GetId(), tc.tagUID, tc.snapshotURI, tc.failure)
+			require.NoError(t, err)
+			require.NoError(t, deleteSession(ctx, client, session.GetId()))
+			assertRetained()
+			_, _, err = client.BeginDeleteSessionCheckpoint(ctx, checkpoint.GetId(), "mallory")
+			require.ErrorIs(t, err, ErrNotFound, "only the owner can delete the checkpoint")
+			_, _, err = client.BeginDeleteSessionCheckpoint(ctx, checkpoint.GetId(), "alice")
+			require.NoError(t, err)
+			assertRetained()
+			// A new store instance must resume a deletion without losing its references.
+			restarted := NewClient(sharedDB, "public")
+			_, tagUID, err := restarted.BeginDeleteSessionCheckpoint(ctx, checkpoint.GetId(), "alice")
+			require.NoError(t, err)
+			require.Equal(t, tc.tagUID, tagUID)
+			assertRetained()
+			require.NoError(t, client.DeleteSessionCheckpoint(ctx, checkpoint.GetId(), "alice"))
+			revisions, err := client.ListUnreferencedRuntimeRevisions(ctx)
+			require.NoError(t, err)
+			require.Len(t, revisions, 1)
 
-	claimed, err := client.BeginRuntimeRevisionDeletion(ctx, "revision")
-	require.NoError(t, err)
-	require.NotNil(t, claimed)
-	require.NoError(t, client.DeleteRuntimeRevision(ctx, "revision", "revision-actor-uid"))
-	_, err = client.GetRuntimeRevision(ctx, "revision")
-	require.ErrorIs(t, err, ErrNotFound)
+			claimed, err := client.BeginRuntimeRevisionDeletion(ctx, "revision")
+			require.NoError(t, err)
+			require.NotNil(t, claimed)
+			require.NoError(t, client.DeleteRuntimeRevision(ctx, "revision", "revision-actor-uid"))
+			_, err = client.GetRuntimeRevision(ctx, "revision")
+			require.ErrorIs(t, err, ErrNotFound)
+		})
+	}
 }
 
 func TestRuntimeRevisionPairReplacement(t *testing.T) {

go/core/internal/service/checkpoint/service.go

@@ -266,13 +266,17 @@ func (s *Service) Delete(ctx context.Context, checkpointID string) error {
 	if err != nil && status.Code(err) != codes.NotFound {
 		return serviceerrors.NewUnavailable("Failed to get checkpoint snapshot tag", err)
 	}
-	if err == nil && (tag.GetMetadata().GetUid() != tagUID ||
-		tag.GetMetadata().GetAtespace() != snapshot.Atespace || tag.GetMetadata().GetName() != tagName(checkpointID) ||
-		tag.GetStatus().GetSnapshot().GetSnapshotUri() != snapshot.URI) {
-		return serviceerrors.NewFailedPrecondition("Checkpoint snapshot tag identity changed", nil)
-	}
-	if err := s.tags.DeleteTag(ctx, snapshot.Atespace, tagName(checkpointID)); err != nil && status.Code(err) != codes.NotFound {
-		return serviceerrors.NewUnavailable("Failed to delete checkpoint snapshot tag", err)
+	if err == nil {
+		// Failed creation records no tag UID after compensation. An unexpected tag
+		// at that name is not ours to delete, even if its snapshot URI matches.
+		if tagUID == "" || tag.GetMetadata().GetUid() != tagUID ||
+			tag.GetMetadata().GetAtespace() != snapshot.Atespace || tag.GetMetadata().GetName() != tagName(checkpointID) ||
+			tag.GetStatus().GetSnapshot().GetSnapshotUri() != snapshot.URI {
+			return serviceerrors.NewFailedPrecondition("Checkpoint snapshot tag identity changed", nil)
+		}
+		if err := s.tags.DeleteTag(ctx, snapshot.Atespace, tagName(checkpointID)); err != nil && status.Code(err) != codes.NotFound {
+			return serviceerrors.NewUnavailable("Failed to delete checkpoint snapshot tag", err)
+		}
 	}
 	if err := s.store.DeleteSessionCheckpoint(ctx, checkpointID, userID); err != nil {
 		return serviceerrors.NewInternal("Failed to delete checkpoint", err)

go/core/internal/service/checkpoint/service_test.go

@@ -317,13 +317,70 @@ func TestDeleteHidesCheckpointBeforeDeletingTag(t *testing.T) {
 		t.Fatal("tag lookup failure must leave deletion pending without deleting the tag")
 	}
 	tags.getErr = nil
+	tags.deleteErr = status.Error(codes.Unavailable, "tag delete unavailable")
+	require.Equal(t, serviceerrors.CodeUnavailable, serviceerrors.CodeOf(service.Delete(ctx, checkpoint.GetId())))
+	require.False(t, store.deleted, "a failed tag delete must retain the database reference")
+	require.NotNil(t, tags.created)
+	tags.deleteErr = nil
 	if err := service.Delete(ctx, checkpoint.GetId()); err != nil {
 		t.Fatal(err)
 	}
-	if checkpoint.State != apiv1alpha1.CheckpointState_CHECKPOINT_STATE_DELETING || tags.deleteCalls != 1 || !store.deleted {
+	if checkpoint.State != apiv1alpha1.CheckpointState_CHECKPOINT_STATE_DELETING || tags.deleteCalls != 2 || !store.deleted {
 		t.Fatalf("checkpoint state = %s, tag deletes = %d, row deleted = %v", checkpoint.State, tags.deleteCalls, store.deleted)
 	}
 }
+func TestDeleteFailedCheckpointWithMissingTag(t *testing.T) {
+	checkpoint := &apiv1alpha1.Checkpoint{Id: "018f47a2-4efb-7c21-a848-123456789abc", State: apiv1alpha1.CheckpointState_CHECKPOINT_STATE_FAILED}
+	store := &testStore{prepared: checkpoint, snapshot: &database.SessionTaskSnapshot{Atespace: "team-a", URI: "s3://snapshots/source"}}
+	// Creation either never made a tag or already compensated it. Once absence is
+	// established, deletion must not issue another unverified delete by name.
+	tags := &testTags{deleteErr: errors.New("unexpected tag deletion")}
+	service := NewService(store, testAuthorizer{}, tags, nil)
+	ctx := auth.AuthSessionTo(t.Context(), testSession{userID: "alice"})
+
+	tags.getErr = status.Error(codes.Unavailable, "tag lookup unavailable")
+	require.Equal(t, serviceerrors.CodeUnavailable, serviceerrors.CodeOf(service.Delete(ctx, checkpoint.GetId())))
+	require.False(t, store.deleted)
+	require.Zero(t, tags.deleteCalls)
+	require.Equal(t, apiv1alpha1.CheckpointState_CHECKPOINT_STATE_DELETING, checkpoint.State)
+
+	// Retry through a new service instance after the backend recovers.
+	tags.getErr = nil
+	service = NewService(store, testAuthorizer{}, tags, nil)
+	require.NoError(t, service.Delete(ctx, checkpoint.GetId()))
+	require.True(t, store.deleted)
+	require.Zero(t, tags.deleteCalls)
+}
+
+func TestDeleteRejectsUnverifiedTag(t *testing.T) {
+	for _, tc := range []struct {
+		name, retainedUID, actualUID string
+	}{
+		{name: "failed checkpoint with unexpected tag", actualUID: "unexpected-uid"},
+		{name: "failed checkpoint with missing tag identity"},
+		{name: "ready checkpoint with replacement tag", retainedUID: "tag-uid", actualUID: "replacement-uid"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			state := apiv1alpha1.CheckpointState_CHECKPOINT_STATE_FAILED
+			if tc.retainedUID != "" {
+				state = apiv1alpha1.CheckpointState_CHECKPOINT_STATE_READY
+			}
+			checkpoint := &apiv1alpha1.Checkpoint{Id: "018f47a2-4efb-7c21-a848-123456789abc", State: state}
+			snapshot := &database.SessionTaskSnapshot{Atespace: "team-a", URI: "s3://snapshots/source"}
+			store := &testStore{prepared: checkpoint, snapshot: snapshot, tagUID: tc.retainedUID}
+			tags := &testTags{created: &ateapipb.Tag{
+				Metadata: &ateapipb.ResourceMetadata{Atespace: snapshot.Atespace, Name: tagName(checkpoint.GetId()), Uid: tc.actualUID},
+				Status:   &ateapipb.TagStatus{Snapshot: &ateapipb.ExternalSnapshot{SnapshotUri: snapshot.URI}},
+			}}
+			service := NewService(store, testAuthorizer{}, tags, nil)
+			ctx := auth.AuthSessionTo(t.Context(), testSession{userID: "alice"})
+			require.Equal(t, serviceerrors.CodeFailedPrecondition, serviceerrors.CodeOf(service.Delete(ctx, checkpoint.GetId())))
+			require.False(t, store.deleted)
+			require.Zero(t, tags.deleteCalls)
+		})
+	}
+}
+
 func TestForkCreatesSessionFromCheckpoint(t *testing.T) {
 	checkpoint := &apiv1alpha1.Checkpoint{Id: "018f47a2-4efb-7c21-a848-123456789abc", State: apiv1alpha1.CheckpointState_CHECKPOINT_STATE_READY}
 	store := &testStore{prepared: checkpoint, snapshot: &database.SessionTaskSnapshot{Atespace: "team-a", URI: "s3://snapshots/snapshot-1", ContentScope: "DATA"}}