feat(store): add structured person profile primitives

cmd/msgvault/cmd/build_cache.go

@@ -481,7 +481,7 @@ type buildResult struct {
 	MaxMessageID  int64
 	OutputDir     string
 	Skipped       bool
-	IdentityOnly  bool // true when only owner_participants/participant_clusters were refreshed
+	IdentityOnly  bool // true when only derived cache datasets were refreshed
 }
 
 // buildCache honors an explicit full rebuild unconditionally. Default builds
@@ -583,12 +583,24 @@ func participantIdentifiersExportSelectSQL() string {
 		FROM sqlite_db.participant_identifiers`
 }
 
+// participantsExportSelectSQL renders the participants dataset export. The
+// full and derived-only builders share this query so participant rows and
+// display names cannot drift between cache publication paths.
+func participantsExportSelectSQL() string {
+	return `SELECT
+			id,
+			COALESCE(TRY_CAST(email_address AS VARCHAR), '') AS email_address,
+			COALESCE(TRY_CAST(domain AS VARCHAR), '') AS domain,
+			COALESCE(TRY_CAST(display_name AS VARCHAR), '') AS display_name,
+			COALESCE(TRY_CAST(phone_number AS VARCHAR), '') AS phone_number
+		FROM sqlite_db.participants`
+}
+
 // derivedDriftOnly reports whether participant-link, conversation-membership,
-// conversation-type, or participant-identifier drift is the only staleness
-// signal. The index-only refresh rebuilds the four relationship datasets from
-// committed base Parquet without re-exporting it (re-staging only the drifted
-// replaceable base dataset: conversation_participants, conversations, or
-// participant_identifiers).
+// conversation-type, participant-identifier, or participant display-name drift
+// is the only staleness signal. The index-only refresh rebuilds the four
+// relationship datasets from committed base Parquet while re-staging any
+// drifted replaceable base dataset.
 //
 // HasAccountIdentityDrift is excluded even though it also bumps
 // identity_revision (and therefore HasIdentityDrift): confirming or
@@ -598,7 +610,8 @@ func participantIdentifiersExportSelectSQL() string {
 // path.
 func derivedDriftOnly(staleness cacheStaleness) bool {
 	return (staleness.HasIdentityDrift || staleness.HasConversationParticipantDrift ||
-		staleness.HasConversationTypeDrift || staleness.HasParticipantIdentifierDrift) &&
+		staleness.HasConversationTypeDrift || staleness.HasParticipantIdentifierDrift ||
+		staleness.HasParticipantDisplayNameDrift) &&
 		!staleness.HasNew && !staleness.HasDeleted &&
 		!staleness.HasUpdated && !staleness.HasAccountIdentityDrift
 }
@@ -678,10 +691,11 @@ func buildCacheLocked(
 	// concurrent identity mutation therefore makes the stamped revision LAG
 	// the store, which HasIdentityDrift detects on the next staleness check —
 	// the cache self-heals. Never move this read after the export. The same
-	// invariant applies to the account-identity revision read alongside it:
-	// this full build derives is_from_me fresh from the current store state,
-	// so stamping a lagging account-identity revision here is likewise
-	// self-healing — HasAccountIdentityDrift catches it on the next check.
+	// invariant applies to the account-identity, participant-identifier, and
+	// participant display-name revisions read alongside it: this full build
+	// derives all of those datasets fresh from the current store state, so
+	// stamping a lagging revision here is likewise self-healing — the matching
+	// staleness check catches it on the next pass.
 	identityStore, err := store.Open(dbPath)
 	if err != nil {
 		return nil, fmt.Errorf("open store for identity export: %w", err)
@@ -701,6 +715,11 @@ func buildCacheLocked(
 		_ = identityStore.Close()
 		return nil, fmt.Errorf("read participant identifier revision: %w", err)
 	}
+	participantDisplayNameRevision, err := identityStore.ParticipantDisplayNameRevision()
+	if err != nil {
+		_ = identityStore.Close()
+		return nil, fmt.Errorf("read participant display-name revision: %w", err)
+	}
 	participantClusters, err := identityStore.ParticipantClusters()
 	if err != nil {
 		_ = identityStore.Close()
@@ -984,18 +1003,12 @@ func buildCacheLocked(
 	escapedParticipantsDir := strings.ReplaceAll(participantsDir, "'", "''")
 	if err := runExport(tableParticipants, fmt.Sprintf(`
 	COPY (
-		SELECT
-			id,
-			COALESCE(TRY_CAST(email_address AS VARCHAR), '') as email_address,
-			COALESCE(TRY_CAST(domain AS VARCHAR), '') as domain,
-			COALESCE(TRY_CAST(display_name AS VARCHAR), '') as display_name,
-			COALESCE(TRY_CAST(phone_number AS VARCHAR), '') as phone_number
-		FROM sqlite_db.participants
+		%s
 	) TO '%s/participants.parquet' (
 		FORMAT PARQUET,
 		COMPRESSION 'zstd'
 	)
-	`, escapedParticipantsDir)); err != nil {
+	`, participantsExportSelectSQL(), escapedParticipantsDir)); err != nil {
 		return nil, fmt.Errorf("export participants: %w", err)
 	}
 
@@ -1364,6 +1377,7 @@ func buildCacheLocked(
 		IdentityRevision:                    identityRevision,
 		AccountIdentityRevision:             accountIdentityRevision,
 		ParticipantIdentifierRevision:       participantIdentifierRevision,
+		ParticipantDisplayNameRevision:      participantDisplayNameRevision,
 		ConversationParticipantsFingerprint: derived.ConversationParticipantsFingerprint,
 		ConversationTypesFingerprint:        typesFingerprint,
 		Stats:                               derived.Stats,

cmd/msgvault/cmd/build_cache_test.go

@@ -3302,7 +3302,8 @@ func TestCacheNeedsBuild_IgnoresAlreadyProcessedUpdatedSyncRun(t *testing.T) {
 // schema version other than the current one now forces a full rebuild.
 func TestCacheNeedsBuild_SchemaVersionMismatch(t *testing.T) {
 	require := require.New(t)
-	require.Equal(17, cacheSchemaVersion, "message_recipients envelope address requires cache v17")
+	require.Equal(18, cacheSchemaVersion,
+		"participant directory revisions require a one-time cache rebuild at v18")
 	tmpDir := setupTestSQLiteEmpty(t)
 
 	dbPath := filepath.Join(tmpDir, "test.db")
@@ -3337,7 +3338,7 @@ func TestCacheNeedsBuild_SchemaVersionMismatch(t *testing.T) {
 	require.False(result.Skipped, "schema mismatch must execute a full rebuild")
 	upgraded, err := query.ReadCacheSyncState(analyticsDir)
 	require.NoError(err, "read upgraded cache state")
-	require.Equal(17, upgraded.SchemaVersion)
+	require.Equal(18, upgraded.SchemaVersion)
 	require.NoFileExists(filepath.Join(analyticsDir, tableParticipantIdentifiers, "data.parquet"),
 		"full rebuild must replace rather than extend the v11 identifier dataset")
 	identifierParquet := filepath.Join(analyticsDir, tableParticipantIdentifiers, "participant_identifiers.parquet")

cmd/msgvault/cmd/cache_derived.go

@@ -84,6 +84,11 @@ func refreshDerivedDatasetsOnly(
 		_ = st.Close()
 		return nil, fmt.Errorf("read participant identifier revision: %w", err)
 	}
+	participantDisplayNameRevision, err := st.ParticipantDisplayNameRevision()
+	if err != nil {
+		_ = st.Close()
+		return nil, fmt.Errorf("read participant display-name revision: %w", err)
+	}
 	clusters, err := st.ParticipantClusters()
 	if err != nil {
 		_ = st.Close()
@@ -139,6 +144,7 @@ func refreshDerivedDatasetsOnly(
 
 	if identityRevision == state.IdentityRevision &&
 		participantIdentifierRevision == state.ParticipantIdentifierRevision &&
+		participantDisplayNameRevision == state.ParticipantDisplayNameRevision &&
 		conversationFingerprint == state.ConversationParticipantsFingerprint &&
 		typesFingerprint == state.ConversationTypesFingerprint {
 		// Nothing the derived datasets read has changed (the account-identity
@@ -177,6 +183,16 @@ func refreshDerivedDatasetsOnly(
 			return nil, err
 		}
 	}
+	displayNamesChanged :=
+		participantDisplayNameRevision != state.ParticipantDisplayNameRevision
+	if identifiersChanged || displayNamesChanged {
+		// Participant identifiers can create participant rows, and display-name
+		// mutations change the row already present in participants.parquet. Both
+		// changes must replace that base dataset before rebuilding the directory.
+		if err := exportDerivedParticipants(ctx, exportDB, staging.root); err != nil {
+			return nil, err
+		}
+	}
 	typesChanged := typesFingerprint != state.ConversationTypesFingerprint
 	if typesChanged {
 		// The index rebuild reads conversation_type from the conversations
@@ -213,11 +229,17 @@ func refreshDerivedDatasetsOnly(
 
 	state.IdentityRevision = identityRevision
 	state.ParticipantIdentifierRevision = participantIdentifierRevision
+	state.ParticipantDisplayNameRevision = participantDisplayNameRevision
 	state.ConversationParticipantsFingerprint = conversationFingerprint
 	state.ConversationTypesFingerprint = typesFingerprint
 	// Stats describe the unchanged committed raw snapshot. Preserve them
 	// byte-for-byte instead of scanning Parquet again.
-	plan := derivedCachePublishPlan(conversationChanged, typesChanged, identifiersChanged)
+	plan := derivedCachePublishPlan(
+		conversationChanged,
+		typesChanged,
+		identifiersChanged,
+		identifiersChanged || displayNamesChanged,
+	)
 	if err := publishDerivedCache(staging, analyticsDir, plan, state, locking); err != nil {
 		return nil, err
 	}
@@ -254,18 +276,19 @@ func fingerprintConversationParticipantsFromSnapshot(
 
 // fingerprintConversationTypesFromSnapshot mirrors
 // sourceConversationTypesFingerprint over the export snapshot, so the stamp
-// written at publish time describes exactly the types the staged datasets
-// baked. The 'email_thread' normalization matches the staleness query (and
-// the CSV snapshot view, which pre-applies it), NOT the exported parquet
-// value — fingerprints only ever compare against each other.
+// written at publish time describes exactly the type/title metadata the staged
+// datasets baked. The normalizations match the staleness query (and the CSV
+// snapshot view), not the exported Parquet values; fingerprints only compare
+// against each other.
 func fingerprintConversationTypesFromSnapshot(
 	ctx context.Context,
 	db sqlRunner,
 	lastMessageID int64,
 ) (string, error) {
 	rows, err := db.QueryContext(ctx, fmt.Sprintf(`
 		SELECT c.id::BIGINT,
-		       COALESCE(TRY_CAST(c.conversation_type AS VARCHAR), 'email_thread')
+		       COALESCE(TRY_CAST(c.conversation_type AS VARCHAR), 'email_thread'),
+		       COALESCE(TRY_CAST(c.title AS VARCHAR), '')
 		FROM sqlite_db.conversations c
 		WHERE EXISTS (
 			SELECT 1
@@ -277,12 +300,12 @@ func fingerprintConversationTypesFromSnapshot(
 		ORDER BY c.id
 	`, exportableMessageWhere("m")), lastMessageID)
 	if err != nil {
-		return "", fmt.Errorf("query conversation types from source snapshot: %w", err)
+		return "", fmt.Errorf("query conversation metadata from source snapshot: %w", err)
 	}
 	defer func() { _ = rows.Close() }()
-	fingerprint, err := identityindex.FingerprintConversationTypes(rows)
+	fingerprint, err := identityindex.FingerprintConversationMetadata(rows)
 	if rowsErr := rows.Err(); rowsErr != nil && err == nil {
-		return "", fmt.Errorf("iterate source conversation types: %w", rowsErr)
+		return "", fmt.Errorf("iterate source conversation metadata: %w", rowsErr)
 	}
 	return fingerprint, err
 }
@@ -346,6 +369,30 @@ func exportDerivedOwnerParticipants(
 	return nil
 }
 
+// exportDerivedParticipants re-stages the participants base dataset when an
+// identifier creates a participant or a display-name mutation changes an
+// existing row. The relationship directory reads this dataset directly.
+func exportDerivedParticipants(
+	ctx context.Context,
+	db sqlRunner,
+	stagingRoot string,
+) error {
+	dir := filepath.Join(stagingRoot, tableParticipants)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return fmt.Errorf("create derived participants directory: %w", err)
+	}
+	path := filepath.Join(dir, "participants.parquet")
+	_, err := db.ExecContext(ctx, fmt.Sprintf(`
+		COPY (
+			%s
+		) TO '%s' (FORMAT PARQUET, COMPRESSION 'zstd')
+	`, participantsExportSelectSQL(), quoteCacheSQL(path)))
+	if err != nil {
+		return fmt.Errorf("export derived participants: %w", err)
+	}
+	return nil
+}
+
 // exportDerivedParticipantIdentifiers re-stages the participant_identifiers
 // base dataset with the full export query so an index-only refresh triggered
 // by identifier drift rebuilds the identity directory from current mappings
@@ -447,7 +494,8 @@ func quoteCacheSQL(value string) string {
 }
 
 func derivedCachePublishPlan(
-	includeConversationParticipants, includeConversations, includeParticipantIdentifiers bool,
+	includeConversationParticipants, includeConversations,
+	includeParticipantIdentifiers, includeParticipants bool,
 ) cachePublishPlan {
 	plan := cachePublishPlan{
 		Append:  make(map[string]bool),
@@ -472,6 +520,9 @@ func derivedCachePublishPlan(
 	if includeParticipantIdentifiers {
 		plan.Replace[tableParticipantIdentifiers] = true
 	}
+	if includeParticipants {
+		plan.Replace[tableParticipants] = true
+	}
 	return plan
 }