Skip to content

fix: scope reprocess/retries to specific batch #304

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
julio4 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: scope reprocess/retries to specific batch #304

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 71 additions & 23 deletions Dayflow/Dayflow/Core/AI/LLMService.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ enum LLMProcessingStep: Sendable, Equatable {

protocol LLMServicing {
func processBatch(
_ batchId: Int64, progressHandler: ((LLMProcessingStep) -> Void)?,
_ batchId: Int64, isReprocess: Bool, progressHandler: ((LLMProcessingStep) -> Void)?,
completion: @escaping (Result<ProcessedBatchResult, Error>) -> Void)
func generateText(prompt: String) async throws -> String
func generateTextStreaming(prompt: String) -> AsyncThrowingStream<String, Error>
Expand Down Expand Up @@ -551,7 +551,8 @@ final class LLMService: LLMServicing {

// Keep the existing processBatch implementation for backward compatibility
func processBatch(
_ batchId: Int64, progressHandler: ((LLMProcessingStep) -> Void)? = nil,
_ batchId: Int64, isReprocess: Bool = false,
progressHandler: ((LLMProcessingStep) -> Void)? = nil,
completion: @escaping (Result<ProcessedBatchResult, Error>) -> Void
) {
Task {
Expand Down Expand Up @@ -689,7 +690,10 @@ final class LLMService: LLMServicing {

// Calculate card-generation lookback window.
let currentTime = Date(timeIntervalSince1970: TimeInterval(batchEndTs))
let windowStartTime = currentTime.addingTimeInterval(-batchingConfig.cardLookbackDuration)
let windowStartTime =
isReprocess
? batchStartDate
: currentTime.addingTimeInterval(-batchingConfig.cardLookbackDuration)

// Fetch observations from the recent batching window (instead of just current batch).
let recentObservations = StorageManager.shared.fetchObservationsByTimeRange(
Expand All @@ -703,25 +707,51 @@ final class LLMService: LLMServicing {
print(" observation: \(obs.observation)")
}

// Fetch existing timeline cards that overlap with the recent batching window.
let existingTimelineCards = StorageManager.shared.fetchTimelineCardsByTimeRange(
from: windowStartTime,
to: currentTime
)

// Convert TimelineCards to ActivityCardData for context
let existingActivityCards = existingTimelineCards.map { card in
ActivityCardData(
startTime: card.startTimestamp,
endTime: card.endTimestamp,
category: card.category,
subcategory: card.subcategory,
title: card.title,
summary: card.summary,
detailedSummary: card.detailedSummary,
distractions: card.distractions,
appSites: card.appSites
// Merge context.
// Live: uses whole lookback window
// Reprocess: uses only immediately-preceding real card
let reprocessMergeAnchor: TimelineCardWithTimestamps? =
isReprocess
? StorageManager.shared.fetchLastTimelineCard(endingBefore: batchStartDate)
.flatMap { $0.title == "Processing failed" ? nil : $0 }
: nil

let existingActivityCards: [ActivityCardData]
if isReprocess {
existingActivityCards =
reprocessMergeAnchor.map { anchor in
[
ActivityCardData(
startTime: anchor.startTimestamp,
endTime: anchor.endTimestamp,
category: anchor.category,
subcategory: anchor.subcategory,
title: anchor.title,
summary: anchor.summary,
detailedSummary: anchor.detailedSummary,
distractions: anchor.distractions,
appSites: nil
)
]
} ?? []
} else {
let existingTimelineCards = StorageManager.shared.fetchTimelineCardsByTimeRange(
from: windowStartTime,
to: currentTime
)
existingActivityCards = existingTimelineCards.map { card in
ActivityCardData(
startTime: card.startTimestamp,
endTime: card.endTimestamp,
category: card.category,
subcategory: card.subcategory,
title: card.title,
summary: card.summary,
detailedSummary: card.detailedSummary,
distractions: card.distractions,
appSites: card.appSites
)
}
}

// Prepare context for activity generation
Expand Down Expand Up @@ -765,12 +795,30 @@ final class LLMService: LLMServicing {
let isBackupGenerated = usedProviderBackup || usedGemmaForCardGeneration
// Note: card generation log is not persisted per-batch yet

// Scope the write
let writeFromTime: Date
let cardsToWrite: [ActivityCardData]
if isReprocess {
let didMerge =
!existingActivityCards.isEmpty && cards.count == existingActivityCards.count
if didMerge, let anchor = reprocessMergeAnchor {
writeFromTime = Date(timeIntervalSince1970: TimeInterval(anchor.startTs))
cardsToWrite = cards
} else {
writeFromTime = batchStartDate
cardsToWrite = Array(cards.dropFirst(existingActivityCards.count))
}
Comment on lines +802 to +810

@praxstack praxstack Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is the one data-loss risk in the PR: didMerge only fires when the model returns exactly 1 card (the anchor path always yields existingActivityCards.count == 1), and every other output shape falls into dropFirst(1), which assumes cards[0] is an unmodified anchor echo. the prompt guarantees neither:

  • GeminiDirectProvider+ActivityCards.swift tells the model previous cards are "a draft you're revising and extending, not locked history", asks "Does your first new card continue the last previous card's work? ... merge them in your output", and says "DEFAULT TO MERGING"
  • validateTimeCoverage is order-insensitive and only checks time coverage, so both traces below pass validation (and the local/backup providers don't run it at all)

two concrete traces:

  1. merge then split (prompt-instructed, common): anchor 10:00-10:15 "coding", retried batch 10:15-10:30 is 7 more min of coding then youtube. model obeys the continuity rules and returns [10:00-10:22 coding (merged), 10:22-10:30 youtube]. count is 2, so didMerge is false, dropFirst(1) drops the merged card, and the write starts at 10:15, so the 10:15-10:22 segment is silently erased. permanent timeline gap.
  2. reorder: model returns [newCard, anchorEcho]. dropFirst(1) drops the real batch card and inserts the anchor echo as a new row while the original anchor row survives outside the delete range, giving a duplicate anchor plus a lost batch card, the same duplication class this PR is fixing.

suggest matching the anchor by time instead of by position/count:

Suggested change
let didMerge =
!existingActivityCards.isEmpty && cards.count == existingActivityCards.count
if didMerge, let anchor = reprocessMergeAnchor {
writeFromTime = Date(timeIntervalSince1970: TimeInterval(anchor.startTs))
cardsToWrite = cards
} else {
writeFromTime = batchStartDate
cardsToWrite = Array(cards.dropFirst(existingActivityCards.count))
}
if let anchor = reprocessMergeAnchor {
// Drop an exact anchor echo wherever it appears (the model may reorder).
var kept = cards
if let echoIndex = kept.firstIndex(where: {
$0.startTime == anchor.startTimestamp && $0.endTime == anchor.endTimestamp
}) {
kept.remove(at: echoIndex)
}
// A card still starting at the anchor's start means the model merged the
// anchor into it: pull the write range back so the stored anchor row is
// replaced instead of left to duplicate.
let didMerge = kept.contains { $0.startTime == anchor.startTimestamp }
writeFromTime =
didMerge
? Date(timeIntervalSince1970: TimeInterval(anchor.startTs))
: batchStartDate
cardsToWrite = kept
} else {
writeFromTime = batchStartDate
cardsToWrite = cards
}

string compare works because the anchor's "h:mm a" timestamps are handed to the model verbatim in the prompt; if you'd rather be strict, parse both sides with the same h:mm a / en_US_POSIX formatter before comparing. worst remaining case (model rewrites the anchor's start time) degrades to an overlapping card rather than a silent gap.

} else {
writeFromTime = windowStartTime
cardsToWrite = cards
}

// Replace old cards with new ones in the time range
let (insertedCardIds, deletedVideoPaths) = StorageManager.shared
.replaceTimelineCardsInRange(
from: windowStartTime,
from: writeFromTime,
to: currentTime,
with: cards.map { card in
with: cardsToWrite.map { card in
TimelineCardShell(
startTimestamp: card.startTime,
endTimestamp: card.endTime,
Expand Down
6 changes: 4 additions & 2 deletions Dayflow/Dayflow/Core/Analysis/AnalysisManager.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ final class AnalysisManager: AnalysisManaging {
)
}

self.queueLLMRequest(batchId: batchId)
self.queueLLMRequest(batchId: batchId, isReprocess: true)

// Wait for batch to complete (check status periodically)
var isCompleted = false
Expand Down Expand Up @@ -371,6 +371,7 @@ final class AnalysisManager: AnalysisManaging {

self.queueLLMRequest(
batchId: batchId,
isReprocess: true,
progressHandler: stepHandler,
completion: { result in
DispatchQueue.main.async {
Expand Down Expand Up @@ -400,6 +401,7 @@ final class AnalysisManager: AnalysisManaging {

private func queueLLMRequest(
batchId: Int64,
isReprocess: Bool = false,
progressHandler: ((LLMProcessingStep) -> Void)? = nil,
completion: ((Result<Void, Error>) -> Void)? = nil
) {
Expand Down Expand Up @@ -473,7 +475,7 @@ final class AnalysisManager: AnalysisManaging {

updateBatchStatus(batchId: batchId, status: "processing")

llmService.processBatch(batchId, progressHandler: progressHandler) {
llmService.processBatch(batchId, isReprocess: isReprocess, progressHandler: progressHandler) {
[weak self] (result: Result<ProcessedBatchResult, Error>) in
guard let self else { return }

Expand Down