fix(sync): discover newly archived threads

docs/commands/sync.md

@@ -43,7 +43,7 @@ discrawl sync --with-media
 
 | Command | Use when | Behavior |
 | --- | --- | --- |
-| `discrawl sync` | routine refresh | skips member refreshes, checks live top-level channels plus active threads, fetches one newest page when no cursor exists, and otherwise fetches only new messages |
+| `discrawl sync` | routine refresh | skips member refreshes, discovers active and newly archived threads, fully indexes new threads, fetches one newest page for other cursorless channels, and otherwise fetches only new messages |
 | `discrawl sync --update=auto` | hybrid Git/live refresh | applies the configured stale snapshot update mode first, then runs the routine live refresh |
 | `discrawl sync --update=force` | intentional exact reconciliation | replaces public snapshot tables first, then runs the routine live refresh |
 | `discrawl sync --all-channels` | repair pass | broad incremental sweep across every stored channel/thread, including archived threads |
@@ -79,7 +79,8 @@ discrawl sync --with-media
 - Retryable failures and unavailable-channel markers are tracked per channel; stale unavailable markers are cleared after a later successful crawl.
 - Marker cleanup is best-effort, so one missing local sync-state row cannot crash the run.
 - Member refresh is best-effort and gives up after five minutes without a caller-supplied deadline. Routine latest-only syncs skip it unless `--with-members` is set.
-- When the archive is already complete, `sync --full` reuses backlog markers and limits steady-state refresh to live top-level channels plus active threads.
+- Routine refreshes keep a per-parent archived-thread cursor, so they discover threads archived between runs without rescanning the historical thread catalog.
+- When the archive is already complete, `sync --full` reuses backlog markers and the same incremental thread discovery instead of revisiting every stored archived thread.
 
 ## See also
 

docs/guides/sync-sources.md

@@ -18,7 +18,7 @@ Sync modes control the Discord bot API side of a run. When `wiretap` is selected
 
 | Command | Use when | Behavior |
 | --- | --- | --- |
-| `discrawl sync` | routine refresh | skips member refreshes, checks live top-level channels plus active threads, fetches one newest page when no cursor exists, and otherwise fetches only new messages |
+| `discrawl sync` | routine refresh | skips member refreshes, discovers active and newly archived threads, fully indexes new threads, fetches one newest page for other cursorless channels, and otherwise fetches only new messages |
 | `discrawl sync --update=auto` | hybrid Git/live refresh | imports a stale Git snapshot first, usually as a changed-shard delta, then runs the routine live refresh |
 | `discrawl sync --all-channels` | repair pass | broad incremental sweep across every stored channel/thread, including archived threads |
 | `discrawl sync --full` | historical backfill | crawls older history until channels are complete; can take a long time on large servers |
@@ -46,7 +46,8 @@ Run one explicit `--full` pass when you want a complete historical guild archive
 - Retryable failures and unavailable-channel markers are tracked per channel; stale unavailable markers are cleared after a later successful crawl.
 - Marker cleanup is best-effort, so one missing local sync-state row cannot crash the run.
 - Full sync member refresh is best-effort and gives up after five minutes without a caller-supplied deadline, so message sync completion is not held hostage by a slow guild member crawl.
-- When the archive is already complete, `sync --full` reuses backlog markers and limits steady-state refresh to live top-level channels plus active threads instead of revisiting every stored archived thread.
+- Routine refreshes keep a per-parent archived-thread cursor, so they discover threads archived between runs without rescanning the historical thread catalog.
+- When the archive is already complete, `sync --full` reuses backlog markers and the same incremental thread discovery instead of revisiting every stored archived thread.
 - If a guild already has a local member snapshot, routine syncs reuse it and skip another full member crawl until that snapshot ages out.
 
 ## See also

internal/cli/cli_test.go

@@ -3646,7 +3646,7 @@ func (f *fakeDiscordClient) GuildThreadsActive(context.Context, string) ([]*disc
 	return nil, nil
 }
 
-func (f *fakeDiscordClient) ThreadsArchived(context.Context, string, bool) ([]*discordgo.Channel, error) {
+func (f *fakeDiscordClient) ThreadsArchived(context.Context, string, bool, time.Time) ([]*discordgo.Channel, error) {
 	return nil, nil
 }
 

internal/discord/client.go

@@ -340,7 +340,7 @@ func (c *Client) GuildThreadsActive(ctx context.Context, guildID string) ([]*dis
 	return list.Threads, nil
 }
 
-func (c *Client) ThreadsArchived(ctx context.Context, channelID string, private bool) ([]*discordgo.Channel, error) {
+func (c *Client) ThreadsArchived(ctx context.Context, channelID string, private bool, after time.Time) ([]*discordgo.Channel, error) {
 	var out []*discordgo.Channel
 	var before *time.Time
 	for {
@@ -359,8 +359,15 @@ func (c *Client) ThreadsArchived(ctx context.Context, channelID string, private
 		if len(list.Threads) == 0 {
 			return out, nil
 		}
-		out = append(out, list.Threads...)
-		if !list.HasMore {
+		reachedAfter := false
+		for _, thread := range list.Threads {
+			if !after.IsZero() && thread != nil && thread.ThreadMetadata != nil && !thread.ThreadMetadata.ArchiveTimestamp.After(after) {
+				reachedAfter = true
+				break
+			}
+			out = append(out, thread)
+		}
+		if reachedAfter || !list.HasMore {
 			return uniqueChannels(out), nil
 		}
 		oldest := list.Threads[len(list.Threads)-1]

internal/discord/client_test.go

@@ -152,11 +152,11 @@ func TestClientRESTWrappers(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, guildActive, 1)
 
-	publicArchived, err := client.ThreadsArchived(ctx, "c1", false)
+	publicArchived, err := client.ThreadsArchived(ctx, "c1", false, time.Time{})
 	require.NoError(t, err)
 	require.Len(t, publicArchived, 1)
 
-	privateArchived, err := client.ThreadsArchived(ctx, "c1", true)
+	privateArchived, err := client.ThreadsArchived(ctx, "c1", true, time.Time{})
 	require.NoError(t, err)
 	require.Len(t, privateArchived, 1)
 
@@ -169,6 +169,54 @@ func TestClientRESTWrappers(t *testing.T) {
 	require.Equal(t, "m1", message.ID)
 }
 
+func TestThreadsArchivedStopsAtAfterCursor(t *testing.T) {
+	after := time.Date(2026, time.August, 19, 12, 0, 0, 0, time.UTC)
+	requests := 0
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v10/channels/c1/threads/archived/public", func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		writeJSON(map[string]any{
+			"threads": []map[string]any{
+				archivedThreadJSON("new", after.Add(time.Minute)),
+				archivedThreadJSON("cursor", after),
+				archivedThreadJSON("old", after.Add(-time.Minute)),
+			},
+			"members":  []any{},
+			"has_more": true,
+		})(w, r)
+	})
+	server := httptest.NewServer(mux)
+	t.Cleanup(server.Close)
+
+	restore := patchDiscordEndpoints(server.URL + "/api/v10/")
+	t.Cleanup(restore)
+	client, err := New("token")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = client.Close() })
+
+	threads, err := client.ThreadsArchived(context.Background(), "c1", false, after)
+	require.NoError(t, err)
+	require.Len(t, threads, 1)
+	require.Equal(t, "new", threads[0].ID)
+	require.Equal(t, 1, requests)
+}
+
+func archivedThreadJSON(id string, archiveAt time.Time) map[string]any {
+	return map[string]any{
+		"id":        id,
+		"guild_id":  "g1",
+		"parent_id": "c1",
+		"name":      id,
+		"type":      11,
+		"thread_metadata": map[string]any{
+			"archived":              true,
+			"auto_archive_duration": 60,
+			"archive_timestamp":     archiveAt.Format(time.RFC3339Nano),
+			"locked":                false,
+		},
+	}
+}
+
 func TestGuildMembersSkipsNilUser(t *testing.T) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/api/v10/guilds/g1/members", writeJSON([]map[string]any{

internal/syncer/channel_catalog.go

@@ -149,8 +149,12 @@ func (s *Syncer) liveChannelList(ctx context.Context, guildID string, mode chann
 	for _, channel := range channels {
 		allChannels[channel.ID] = channel
 	}
+	parentIDs := scopedThreadParentIDs(channels, exclusions)
 	if mode == channelCatalogIncremental {
-		if err := s.appendActiveThreadCatalog(ctx, allChannels, guildID, scopedThreadParentIDs(channels, exclusions)); err != nil {
+		if err := s.appendActiveThreadCatalog(ctx, allChannels, guildID, parentIDs); err != nil {
+			return nil, err
+		}
+		if err := s.appendIncrementalArchivedThreadCatalog(ctx, allChannels, parentIDs); err != nil {
 			return nil, err
 		}
 		return mapsToSlice(allChannels), nil
@@ -164,7 +168,6 @@ func (s *Syncer) liveChannelList(ctx context.Context, guildID string, mode chann
 		storedRows = rows
 		mergeStoredThreadChannels(allChannels, rows)
 	}
-	parentIDs := scopedThreadParentIDs(channels, exclusions)
 	if len(storedThreadParentIDs(storedRows)) == 0 {
 		if err := s.appendThreadCatalog(ctx, allChannels, parentIDs); err != nil {
 			return nil, err
@@ -202,19 +205,44 @@ func (s *Syncer) appendThreadCatalog(ctx context.Context, allChannels map[string
 			return err
 		}
 		failed := unavailable
-		for _, private := range []bool{false, true} {
-			archived, err := s.client.ThreadsArchived(ctx, channel.ID, private)
+		for _, private := range archivedThreadPrivacy(channel) {
+			if s.appendArchivedThreads(ctx, allChannels, channel.ID, private, time.Time{}) {
+				failed = true
+			}
+		}
+		if !failed {
+			if err := s.clearThreadCatalogUnavailableChannel(ctx, channel.ID); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (s *Syncer) appendIncrementalArchivedThreadCatalog(ctx context.Context, allChannels map[string]*discordgo.Channel, parents []string) error {
+	scanStartedAt := time.Now().UTC()
+	initialCursor, err := s.archivedThreadInitialCursor(ctx, scanStartedAt)
+	if err != nil {
+		return err
+	}
+	for _, parentID := range uniqueIDs(parents) {
+		channel := allChannels[parentID]
+		if !isThreadParent(channel) {
+			continue
+		}
+		failed := false
+		for _, private := range archivedThreadPrivacy(channel) {
+			scope := channelArchivedThreadCursorScope(channel.ID, private)
+			after, err := s.archivedThreadCursor(ctx, scope, initialCursor)
 			if err != nil {
-				if s.skipThreadCatalogUnavailableChannelByID(ctx, channel.ID, err, "thread archive crawl failed") {
-					failed = true
-					continue
-				}
-				s.logger.Warn("thread archive crawl failed", "channel_id", channel.ID, "private", private, "err", err)
+				return err
+			}
+			if s.appendArchivedThreads(ctx, allChannels, channel.ID, private, after) {
 				failed = true
 				continue
 			}
-			for _, thread := range archived {
-				allChannels[thread.ID] = thread
+			if err := s.store.SetSyncState(ctx, scope, scanStartedAt.Format(time.RFC3339Nano)); err != nil {
+				return err
 			}
 		}
 		if !failed {
@@ -226,6 +254,60 @@ func (s *Syncer) appendThreadCatalog(ctx context.Context, allChannels map[string
 	return nil
 }
 
+func (s *Syncer) archivedThreadInitialCursor(ctx context.Context, fallback time.Time) (time.Time, error) {
+	raw, err := s.store.GetSyncState(ctx, "sync:last_success")
+	if err != nil {
+		return time.Time{}, err
+	}
+	if raw == "" {
+		return fallback, nil
+	}
+	parsed, err := time.Parse(time.RFC3339Nano, raw)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("parse last successful sync time: %w", err)
+	}
+	return parsed, nil
+}
+
+func (s *Syncer) appendArchivedThreads(ctx context.Context, allChannels map[string]*discordgo.Channel, channelID string, private bool, after time.Time) bool {
+	archived, err := s.client.ThreadsArchived(ctx, channelID, private, after)
+	if err != nil {
+		if !s.skipThreadCatalogUnavailableChannelByID(ctx, channelID, err, "thread archive crawl failed") {
+			s.logger.Warn("thread archive crawl failed", "channel_id", channelID, "private", private, "err", err)
+		}
+		return true
+	}
+	for _, thread := range archived {
+		allChannels[thread.ID] = thread
+	}
+	return false
+}
+
+func (s *Syncer) archivedThreadCursor(ctx context.Context, scope string, initial time.Time) (time.Time, error) {
+	raw, err := s.store.GetSyncState(ctx, scope)
+	if err != nil {
+		return time.Time{}, err
+	}
+	if raw == "" {
+		if err := s.store.SetSyncState(ctx, scope, initial.Format(time.RFC3339Nano)); err != nil {
+			return time.Time{}, err
+		}
+		return initial, nil
+	}
+	parsed, err := time.Parse(time.RFC3339Nano, raw)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("parse archived thread cursor %s: %w", scope, err)
+	}
+	return parsed, nil
+}
+
+func archivedThreadPrivacy(channel *discordgo.Channel) []bool {
+	if channel.Type == discordgo.ChannelTypeGuildText {
+		return []bool{false, true}
+	}
+	return []bool{false}
+}
+
 func (s *Syncer) appendActiveThreadCatalog(ctx context.Context, allChannels map[string]*discordgo.Channel, guildID string, parents []string) error {
 	allowedParents := make(map[string]struct{}, len(parents))
 	for _, parentID := range uniqueIDs(parents) {

internal/syncer/channel_catalog_test.go

@@ -691,7 +691,7 @@ func TestFullSyncUsesGuildActiveThreadsForStoredParents(t *testing.T) {
 	_, err = svc.Sync(ctx, SyncOptions{Full: true, GuildIDs: []string{"g1"}})
 	require.NoError(t, err)
 	require.Equal(t, 1, client.guildThreadCalls)
-	require.Zero(t, client.threadCalls)
+	require.Equal(t, 2, client.threadCalls)
 }
 
 func TestFullSyncDiscoversActiveThreadUnderNewParent(t *testing.T) {
@@ -772,7 +772,7 @@ func TestFullSyncDiscoversActiveThreadUnderNewParent(t *testing.T) {
 	require.Equal(t, 2, stats.Threads)
 	require.Equal(t, 1, stats.Messages)
 	require.Equal(t, 1, client.guildThreadCalls)
-	require.Zero(t, client.threadCalls)
+	require.Equal(t, 4, client.threadCalls)
 	require.Equal(t, 1, client.messageCalls["t2"])
 
 	results, err := s.SearchMessages(ctx, store.SearchOptions{Query: "new parent thread"})

internal/syncer/message_sync.go

@@ -237,6 +237,9 @@ func (s *Syncer) syncChannelMessages(ctx context.Context, guildID string, channe
 		return 0, nil
 	}
 	if latestOnly {
+		if isThreadChannel(channel) && !state.BackfillComplete {
+			return s.syncFullChannelHistory(ctx, channel, state, embeddings, since, progress)
+		}
 		if state.Latest == "" {
 			return s.syncLatestChannelHistory(ctx, channel, embeddings, since, progress)
 		}

internal/syncer/records.go

@@ -220,6 +220,14 @@ func channelThreadCatalogUnavailableScope(channelID string) string {
 	return "channel:" + channelID + ":thread_catalog_unavailable"
 }
 
+func channelArchivedThreadCursorScope(channelID string, private bool) string {
+	kind := "public"
+	if private {
+		kind = "private"
+	}
+	return "channel:" + channelID + ":archived_" + kind + "_threads_after"
+}
+
 func makeGuildSet(ids []string) map[string]struct{} {
 	if len(ids) == 0 {
 		return nil

internal/syncer/syncer.go

@@ -22,7 +22,7 @@ type Client interface {
 	GuildChannels(context.Context, string) ([]*discordgo.Channel, error)
 	ThreadsActive(context.Context, string) ([]*discordgo.Channel, error)
 	GuildThreadsActive(context.Context, string) ([]*discordgo.Channel, error)
-	ThreadsArchived(context.Context, string, bool) ([]*discordgo.Channel, error)
+	ThreadsArchived(context.Context, string, bool, time.Time) ([]*discordgo.Channel, error)
 	GuildMembers(context.Context, string) ([]*discordgo.Member, error)
 	ChannelMessages(context.Context, string, int, string, string) ([]*discordgo.Message, error)
 	ChannelMessage(context.Context, string, string) (*discordgo.Message, error)