feat(artifact): import normalized peer checkpoints (#1299)

Artifact export currently establishes durable publication authority, but artifacts already present in a store cannot be consumed into the normalized archive with equivalent crash safety. That leaves round-trip synchronization one-way and makes later folder or service transports unsafe to add because inbound work has no durable claim, version-deferral, or landing boundary.

This change adds exact checkpoint claims, monotonic peer heads, complete landing maps, and per-session provenance that survive full resync. Imports authenticate bounded artifact reads, accept semantically valid peer JSON under its stored identity, defer missing or future dependencies independently, quarantine deterministic corruption, and rewrite complete sessions under origin-qualified identities without carrying source-machine or unverified secret state.

The coordinator acknowledges a checkpoint only after its available session closures and exact landing map are durable. Retries converge across each write boundary, unchanged large checkpoints perform closure work only for changed sessions, and repeated observations do not duplicate normalized messages or usage.

Checkpoint absence remains non-destructive. Metadata replay, artifact transport, raw provider-source stewardship, and provider-file eviction remain separate follow-on scopes.

Co-authored-by: Wes McKinney <wesm@users.noreply.github.com>

Author: wesm

internal/artifact/export.go

@@ -13,6 +13,7 @@ import (
 	"io/fs"
 	"os"
 	"sort"
+	"strings"
 
 	"go.kenn.io/agentsview/internal/db"
 )
@@ -407,6 +408,9 @@ func exportFullToStoreWithDrainRoundsAndLimits(
 		if _, _, err := exportClaimedSessionToStore(
 			ctx, database, store, origin, sess, limits,
 		); err != nil {
+			if isDeterministicArtifactExportError(err) {
+				continue
+			}
 			return result, err
 		}
 	}
@@ -502,6 +506,12 @@ func exportLoadedSessionToStore(
 	usageEvents []db.UsageEvent,
 	limits artifactLimits,
 ) (string, bool, error) {
+	if sess.ID == "" || strings.Contains(sess.ID, "~") {
+		return "", false, rejectArtifactExportf(
+			"native session ID %q is invalid: must be non-empty and exclude '~'",
+			sess.ID,
+		)
+	}
 	if len(messages) > limits.sessionMessages {
 		return "", false, rejectArtifactExportf(
 			"session message limit exceeded for %s: got %d, limit %d",

internal/artifact/export_limits_test.go

@@ -112,6 +112,46 @@ func TestExportClassifiesBoundedLoadLimit(t *testing.T) {
 	assertNoPublishedArtifacts(t, store, contractOrigin)
 }
 
+func TestExportRejectsNativeSessionIDsImporterCannotRepresent(t *testing.T) {
+	tests := []struct {
+		name      string
+		sessionID string
+	}{
+		{name: "empty", sessionID: ""},
+		{name: "reserved separator", sessionID: "native~session"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			store := newTestArtifactStore(t)
+			session := &db.Session{
+				ID:               tt.sessionID,
+				Machine:          "local",
+				MessageCount:     1,
+				UserMessageCount: 1,
+			}
+			messages := []db.Message{{
+				SessionID: tt.sessionID,
+				Ordinal:   0,
+				Role:      "user",
+				Content:   "hello",
+			}}
+
+			_, _, err := exportLoadedSessionToStore(
+				t.Context(),
+				store,
+				contractOrigin,
+				session,
+				messages,
+				nil,
+				productionArtifactLimits(),
+			)
+			require.ErrorIs(t, err, ErrArtifactExportRejected)
+			assert.Contains(t, err.Error(), "native session ID")
+			assertNoPublishedArtifacts(t, store, contractOrigin)
+		})
+	}
+}
+
 func TestExportRejectsNestedAmplificationBeforePublication(t *testing.T) {
 	tests := []struct {
 		name      string

internal/artifact/export_test.go

@@ -468,6 +468,108 @@ func TestExportContinuesPastInvalidManifestValue(t *testing.T) {
 	assert.Contains(t, rejection.Error, "encoding manifest for invalid")
 }
 
+func TestFullExportSkipsPreviouslyRejectedNativeSessionID(t *testing.T) {
+	database := testExportDB(t)
+	store := newTestArtifactStore(t)
+	seedSession(t, database, "invalid~native", "alpha")
+
+	rejected, err := ExportToStore(
+		t.Context(), database, store, ExportOptions{Origin: contractOrigin},
+	)
+	require.NoError(t, err)
+	require.Equal(t, 1, rejected.RejectedSessions)
+
+	result, err := ExportToStore(
+		t.Context(), database, store, ExportOptions{
+			Origin: contractOrigin,
+			Full:   true,
+		},
+	)
+	require.NoError(t, err)
+	assert.Zero(t, result.ExportedSessions)
+	assert.Zero(t, result.RejectedSessions)
+	pending, err := database.PendingArtifactExports(t.Context(), 10)
+	require.NoError(t, err)
+	assert.Empty(t, pending)
+}
+
+func TestFullExportRemovesPublishedNativeSessionIDRejectedByCurrentWire(t *testing.T) {
+	ctx := t.Context()
+	databasePath := filepath.Join(t.TempDir(), "archive.db")
+	database, err := db.Open(databasePath)
+	require.NoError(t, err)
+	require.NoError(t, database.SetSyncState(originStateKey, contractOrigin))
+	store := newTestArtifactStore(t)
+	seedSession(t, database, "legacy~native", "alpha")
+
+	claims, err := database.ArtifactExportClaims(ctx, []string{"legacy~native"})
+	require.NoError(t, err)
+	require.Len(t, claims, 1)
+	manifestHash := strings.Repeat("a", 64)
+	revision, changed, err := database.ApplyArtifactPublicationChanges(
+		ctx, contractOrigin, []db.ArtifactPublicationChange{{
+			SessionID:         "legacy~native",
+			Generation:        claims[0].Generation,
+			ManifestHash:      manifestHash,
+			SourceFingerprint: manifestHash,
+		}},
+	)
+	require.NoError(t, err)
+	require.True(t, changed)
+	publications, mapDigest, mapRevision, err := spoolArtifactPublicationMap(
+		ctx, database, contractOrigin,
+	)
+	require.NoError(t, err)
+	require.Equal(t, revision, mapRevision)
+	checkpointBody, checkpointIdentity, err := spoolArtifactCheckpoint(
+		ctx, publications, contractOrigin, 1,
+	)
+	require.NoError(t, err)
+	checkpointRef := requireContractRef(
+		t, contractOrigin, KindCheckpoints, "cp-0000000001.json",
+	)
+	_, err = store.Create(
+		ctx, checkpointRef, checkpointIdentity,
+		canonicalArtifactMediaType(KindCheckpoints), checkpointBody,
+	)
+	require.NoError(t, err)
+	require.NoError(t, closeAndRemoveExportSpool(publications))
+	require.NoError(t, closeAndRemoveExportSpool(checkpointBody))
+	require.NoError(t, database.RecordArtifactCheckpointHead(
+		ctx, db.ArtifactCheckpointHead{
+			Origin:              contractOrigin,
+			Sequence:            1,
+			PublicationRevision: revision,
+			SessionMapSHA256:    mapDigest,
+			CheckpointSHA256:    checkpointIdentity.SHA256,
+			CheckpointSize:      checkpointIdentity.Size,
+		},
+		claims,
+	))
+	require.NoError(t, database.Close())
+
+	database, err = db.Open(databasePath)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, database.Close()) })
+	result, err := ExportToStore(ctx, database, store, ExportOptions{
+		Origin: contractOrigin,
+		Full:   true,
+	})
+	require.NoError(t, err)
+	assert.Equal(t, 1, result.RejectedSessions)
+	published := latestStoreCheckpointForTest(t, store, contractOrigin)
+	assert.NotContains(t, published.Sessions, contractOrigin+"~legacy~native")
+	pending, err := database.PendingArtifactExports(ctx, 10)
+	require.NoError(t, err)
+	assert.Empty(t, pending)
+	rejection, ok, err := database.GetArtifactExportRejection(
+		ctx, "legacy~native",
+	)
+	require.NoError(t, err)
+	require.True(t, ok)
+	assert.Contains(t, rejection.Error, "native session ID")
+}
+
 func TestExportTransientFailureDoesNotReject(t *testing.T) {
 	database := testExportDB(t)
 	seedSession(t, database, "rejected", "alpha")