This commit is contained in:
2 files changed
+202
-5
No files matched your search
@@ -213,9 +213,16 @@ class ChatSubagentsDelegate(
|
|||||||
* Rules:
|
* Rules:
|
||||||
* - `subagent_id` is the primary identity.
|
* - `subagent_id` is the primary identity.
|
||||||
* - Protects against races: if an indicator received a push event with timestamp > [requestTime],
|
* - Protects against races: if an indicator received a push event with timestamp > [requestTime],
|
||||||
* its push-event status/type is preserved.
|
* its push-event status/type is preserved and it is not pruned.
|
||||||
* - Preserves local details (logs, taskIndex, taskCount, summary) not provided in the snapshot.
|
* - Preserves local details (logs, taskIndex, taskCount, summary) not provided in the snapshot.
|
||||||
* - Does not discard locally known indicators absent from the snapshot.
|
* - Reconciles stale running indicators (issue #1117): an identified ([SubagentIndicator.subagentId] != null),
|
||||||
|
* running ([SubagentIndicator.isRunning]) indicator omitted from an authoritative snapshot is removed
|
||||||
|
* if its [SubagentIndicator.lastEventTimestamp] <= [requestTime]. This clears indicators that completed
|
||||||
|
* while disconnected where `subagent.complete` was missed.
|
||||||
|
* - Protects against races for new subagents: a running indicator spawned or updated by a push event
|
||||||
|
* newer than [requestTime] is preserved even if omitted from an older in-flight snapshot response.
|
||||||
|
* - Preserves already-terminal indicators (completed, failed, cancelled) and unassigned placeholders
|
||||||
|
* ([SubagentIndicator.subagentId] == null) even if omitted from the snapshot.
|
||||||
*/
|
*/
|
||||||
internal fun mergeSubagentList(
|
internal fun mergeSubagentList(
|
||||||
current: List<SubagentIndicator>,
|
current: List<SubagentIndicator>,
|
||||||
@@ -300,6 +307,15 @@ class ChatSubagentsDelegate(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
val snapshotIds = listItems.mapNotNull { it.subagentId.trim().takeIf { id -> id.isNotEmpty() } }.toSet()
|
||||||
|
merged.removeAll { indicator ->
|
||||||
|
val id = indicator.subagentId?.trim()
|
||||||
|
!id.isNullOrEmpty() &&
|
||||||
|
indicator.isRunning &&
|
||||||
|
id !in snapshotIds &&
|
||||||
|
indicator.lastEventTimestamp <= requestTime
|
||||||
|
}
|
||||||
|
|
||||||
return merged
|
return merged
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -218,7 +218,7 @@ class ChatSubagentsDelegateTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun testHydrateSubagents_preservesIndicatorsOmittedFromSnapshot() =
|
fun testHydrateSubagents_removesStaleRunningIndicatorOmittedFromSnapshot() =
|
||||||
testScope.runTest {
|
testScope.runTest {
|
||||||
val requestTime = 1000L
|
val requestTime = 1000L
|
||||||
val current =
|
val current =
|
||||||
@@ -228,12 +228,14 @@ class ChatSubagentsDelegateTest {
|
|||||||
subagentId = "sub-1",
|
subagentId = "sub-1",
|
||||||
goal = "Active worker 1",
|
goal = "Active worker 1",
|
||||||
status = "running",
|
status = "running",
|
||||||
|
lastEventTimestamp = 900L,
|
||||||
),
|
),
|
||||||
SubagentIndicator(
|
SubagentIndicator(
|
||||||
type = "subagent.progress",
|
type = "subagent.progress",
|
||||||
subagentId = "sub-2",
|
subagentId = "sub-2",
|
||||||
goal = "Active worker 2",
|
goal = "Active worker 2",
|
||||||
status = "running",
|
status = "running",
|
||||||
|
lastEventTimestamp = 900L,
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -247,11 +249,190 @@ class ChatSubagentsDelegateTest {
|
|||||||
)
|
)
|
||||||
|
|
||||||
val result = delegate.mergeSubagentList(current, listItems, requestTime)
|
val result = delegate.mergeSubagentList(current, listItems, requestTime)
|
||||||
assertEquals(2, result.size)
|
assertEquals(1, result.size)
|
||||||
assertTrue(result.any { it.subagentId == "sub-1" })
|
assertFalse(result.any { it.subagentId == "sub-1" })
|
||||||
assertTrue(result.any { it.subagentId == "sub-2" })
|
assertTrue(result.any { it.subagentId == "sub-2" })
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testHydrateSubagents_newerPushEventIndicatorOmittedFromSnapshotIsNotRemoved() =
|
||||||
|
testScope.runTest {
|
||||||
|
val requestTime = 1000L
|
||||||
|
val current =
|
||||||
|
listOf(
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.start",
|
||||||
|
subagentId = "sub-1",
|
||||||
|
goal = "Spawned during request",
|
||||||
|
status = "running",
|
||||||
|
lastEventTimestamp = 1500L, // Newer than requestTime (1000L)
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
// Older snapshot in-flight returns without sub-1
|
||||||
|
val listItems = emptyList<SubagentListItem>()
|
||||||
|
|
||||||
|
val result = delegate.mergeSubagentList(current, listItems, requestTime)
|
||||||
|
assertEquals(1, result.size)
|
||||||
|
assertEquals("sub-1", result.first().subagentId)
|
||||||
|
assertEquals("running", result.first().status)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testHydrateSubagents_placeholderWithoutSubagentIdIsPreserved() =
|
||||||
|
testScope.runTest {
|
||||||
|
val requestTime = 1000L
|
||||||
|
val current =
|
||||||
|
listOf(
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.spawn_requested",
|
||||||
|
subagentId = null,
|
||||||
|
goal = "Plan decomposition",
|
||||||
|
status = "running",
|
||||||
|
lastEventTimestamp = 800L,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
val listItems = emptyList<SubagentListItem>()
|
||||||
|
|
||||||
|
val result = delegate.mergeSubagentList(current, listItems, requestTime)
|
||||||
|
assertEquals(1, result.size)
|
||||||
|
assertNull(result.first().subagentId)
|
||||||
|
assertEquals("Plan decomposition", result.first().goal)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testHydrateSubagents_terminalIndicatorsOmittedFromSnapshotArePreserved() =
|
||||||
|
testScope.runTest {
|
||||||
|
val requestTime = 2000L
|
||||||
|
val current =
|
||||||
|
listOf(
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.complete",
|
||||||
|
subagentId = "sub-completed",
|
||||||
|
goal = "Finished task",
|
||||||
|
status = "completed",
|
||||||
|
lastEventTimestamp = 1000L,
|
||||||
|
),
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.progress",
|
||||||
|
subagentId = "sub-failed",
|
||||||
|
goal = "Failed task",
|
||||||
|
status = "failed",
|
||||||
|
lastEventTimestamp = 1000L,
|
||||||
|
),
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.progress",
|
||||||
|
subagentId = "sub-cancelled",
|
||||||
|
goal = "Cancelled task",
|
||||||
|
status = "cancelled",
|
||||||
|
lastEventTimestamp = 1000L,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
// Snapshot active roster omits all terminal subagents
|
||||||
|
val listItems = emptyList<SubagentListItem>()
|
||||||
|
|
||||||
|
val result = delegate.mergeSubagentList(current, listItems, requestTime)
|
||||||
|
assertEquals(3, result.size)
|
||||||
|
assertTrue(result.any { it.subagentId == "sub-completed" && it.isComplete })
|
||||||
|
assertTrue(result.any { it.subagentId == "sub-failed" && it.isFailed })
|
||||||
|
assertTrue(result.any { it.subagentId == "sub-cancelled" && it.isCancelled })
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testHydrateSubagents_nullResponseDoesNotClearIndicators() =
|
||||||
|
testScope.runTest {
|
||||||
|
val initial =
|
||||||
|
listOf(
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.progress",
|
||||||
|
subagentId = "sub-1",
|
||||||
|
goal = "Active worker",
|
||||||
|
status = "running",
|
||||||
|
lastEventTimestamp = 500L,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
uiState.value = uiState.value.copy(subagentIndicators = initial)
|
||||||
|
|
||||||
|
coEvery { mockRepository.listSubagents("session-123") } returns null
|
||||||
|
|
||||||
|
delegate.hydrateSubagents("session-123")
|
||||||
|
advanceUntilIdle()
|
||||||
|
|
||||||
|
assertEquals(initial, uiState.value.subagentIndicators)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testHydrateSubagents_exceptionDoesNotClearIndicators() =
|
||||||
|
testScope.runTest {
|
||||||
|
val initial =
|
||||||
|
listOf(
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.progress",
|
||||||
|
subagentId = "sub-1",
|
||||||
|
goal = "Active worker",
|
||||||
|
status = "running",
|
||||||
|
lastEventTimestamp = 500L,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
uiState.value = uiState.value.copy(subagentIndicators = initial)
|
||||||
|
|
||||||
|
coEvery { mockRepository.listSubagents("session-123") } throws RuntimeException("Network timeout")
|
||||||
|
|
||||||
|
delegate.hydrateSubagents("session-123")
|
||||||
|
advanceUntilIdle()
|
||||||
|
|
||||||
|
assertEquals(initial, uiState.value.subagentIndicators)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testHydrateSubagents_reconnectReconciliationRemovesStaleRunningCard() =
|
||||||
|
testScope.runTest {
|
||||||
|
uiState.value =
|
||||||
|
uiState.value.copy(
|
||||||
|
subagentIndicators =
|
||||||
|
listOf(
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.start",
|
||||||
|
subagentId = "sub-1",
|
||||||
|
goal = "Download files",
|
||||||
|
status = "running",
|
||||||
|
lastEventTimestamp = 500L,
|
||||||
|
),
|
||||||
|
SubagentIndicator(
|
||||||
|
type = "subagent.start",
|
||||||
|
subagentId = "sub-2",
|
||||||
|
goal = "Process files",
|
||||||
|
status = "running",
|
||||||
|
lastEventTimestamp = 500L,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
// sub-1 finished while disconnected, sub-2 is still running
|
||||||
|
coEvery { mockRepository.listSubagents("session-123") } returns
|
||||||
|
SubagentListResponse(
|
||||||
|
subagents =
|
||||||
|
listOf(
|
||||||
|
SubagentListItem(
|
||||||
|
subagentId = "sub-2",
|
||||||
|
goal = "Process files",
|
||||||
|
status = "running",
|
||||||
|
elapsedSeconds = 15.0,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
delegate.hydrateSubagents("session-123")
|
||||||
|
advanceUntilIdle()
|
||||||
|
|
||||||
|
val indicators = uiState.value.subagentIndicators
|
||||||
|
assertEquals(1, indicators.size)
|
||||||
|
assertEquals("sub-2", indicators.first().subagentId)
|
||||||
|
assertFalse(indicators.any { it.subagentId == "sub-1" })
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun testInspectSubagentTranscript_replacesTailWithoutDuplicateAppending() =
|
fun testInspectSubagentTranscript_replacesTailWithoutDuplicateAppending() =
|
||||||
testScope.runTest {
|
testScope.runTest {
|
||||||
|
|||||||
Reference in new issue
Block a user