From e7eb73bf53e4103203e2347779ae6441a5cbc6f7 Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Mon, 17 Aug 2026 18:48:34 +0100 Subject: [PATCH] Preserve reconstructed and cancelled turn states Signed-off-by: kenny lopez --- crates/buzz-acp/src/pool.rs | 47 +++++++++++--- docs/nips/NIP-AO.md | 10 +++ .../agent_activity/active_agent_turns.dart | 33 ++++++++-- .../composer_agent_activity_indicator.dart | 11 +++- .../agent_activity_controls.dart | 12 +++- .../active_agent_turns_test.dart | 62 +++++++++++++++++++ ...omposer_agent_activity_indicator_test.dart | 47 ++++++++++++++ 7 files changed, 204 insertions(+), 18 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 45a1f2fbe..2fe32b538 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -1529,7 +1529,7 @@ pub async fn run_prompt_task( // metadata now, before the agent is moved into PromptResult. It must be // declared before `liveness_guard`: Rust drops locals in reverse order, so // liveness is aborted before completion makes the turn terminal. - let _turn_guard = TurnCompletionGuard::new( + let mut turn_guard = TurnCompletionGuard::new( agent.acp.observer_handle(), agent.acp.observer_agent_index(), observer_channel_id, @@ -2245,6 +2245,7 @@ pub async fn run_prompt_task( { Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); + turn_guard.mark_cancelled(); agent.state.invalidate(&source); let retry_batch = requeue_cancelled_batch(&ctx, control_signal, batch); @@ -2376,6 +2377,9 @@ pub async fn run_prompt_task( match prompt_result { Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); + if matches!(&stop_reason, StopReason::Cancelled) { + turn_guard.mark_cancelled(); + } if let PromptSource::Channel(cid) = &source { let standing_sent = !agent.has_system_prompt_support(); @@ -3994,6 +3998,7 @@ struct TurnCompletionGuard { channel_id: Option, thread_head_id: Option, turn_id: String, + cancelled: bool, } impl TurnCompletionGuard { @@ -4010,8 +4015,13 @@ impl TurnCompletionGuard { channel_id, thread_head_id, turn_id, + cancelled: false, } } + + fn mark_cancelled(&mut self) { + self.cancelled = true; + } } impl Drop for TurnCompletionGuard { @@ -4020,12 +4030,12 @@ impl Drop for TurnCompletionGuard { let mut context = observer::context_for(self.channel_id, None, Some(self.turn_id.clone())); context.thread_head_id = self.thread_head_id.clone(); - observer.emit( - "turn_completed", - self.agent_index, - &context, - serde_json::json!({}), - ); + let payload = if self.cancelled { + serde_json::json!({ "outcome": "cancelled" }) + } else { + serde_json::json!({}) + }; + observer.emit("turn_completed", self.agent_index, &context, payload); } } } @@ -6815,6 +6825,29 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" .find(|event| event.kind == "turn_completed") .expect("completion frame"); assert_eq!(event.thread_head_id.as_deref(), Some("thread-1")); + assert_eq!(event.payload, serde_json::json!({})); + } + + #[test] + fn test_completion_guard_reports_cancelled_outcome() { + let observer = observer::ObserverHandle::in_process(); + { + let mut guard = TurnCompletionGuard::new( + Some(observer.clone()), + Some(0), + Some(Uuid::new_v4()), + Some("thread-1".into()), + "turn-1".into(), + ); + guard.mark_cancelled(); + } + + let event = observer + .snapshot() + .into_iter() + .find(|event| event.kind == "turn_completed") + .expect("completion frame"); + assert_eq!(event.payload, serde_json::json!({ "outcome": "cancelled" })); } #[tokio::test(start_paused = true)] diff --git a/docs/nips/NIP-AO.md b/docs/nips/NIP-AO.md index 340506986..553f8d521 100644 --- a/docs/nips/NIP-AO.md +++ b/docs/nips/NIP-AO.md @@ -111,8 +111,18 @@ Unknown `kind` values MUST be ignored. | `acp_read` | Inbound ACP protocol frame (model → harness) | | `acp_write` | Outbound ACP protocol frame (harness → model) | | `turn_started` | A new agent turn has begun | +| `turn_liveness` | The current turn is still active | +| `turn_completed` | The current turn ended, optionally with an outcome | +| `turn_error` | The current turn stopped with an error | +| `agent_panic` | The agent task terminated unexpectedly | | `session_resolved` | Session completed or terminated | +`turn_completed.payload.outcome` MAY be `"cancelled"` when the turn was +explicitly cancelled. A missing outcome retains the legacy completion semantics +and SHOULD be treated as finished unless a more specific terminal frame exists. +`turn_started` and `turn_liveness` carry `livenessIntervalSecs`; other mid-turn +frames are not required to repeat it. + ### Control (`frame=control`) The `content` field decrypts to: diff --git a/mobile/lib/features/channels/agent_activity/active_agent_turns.dart b/mobile/lib/features/channels/agent_activity/active_agent_turns.dart index f34181f71..7de03414a 100644 --- a/mobile/lib/features/channels/agent_activity/active_agent_turns.dart +++ b/mobile/lib/features/channels/agent_activity/active_agent_turns.dart @@ -14,7 +14,7 @@ const _maximumTurnDuration = Duration(days: 7); const _livenessTimeoutSlack = Duration(seconds: 30); /// Lifecycle state reconstructed from owner-scoped observer frames. -enum AgentTurnPhase { working, finished, error } +enum AgentTurnPhase { working, finished, cancelled, error } /// One observed agent turn, including its explicit terminal outcome when known. @immutable @@ -116,7 +116,7 @@ List reduceAgentTurnStates( case 'turn_error': case 'agent_panic': final terminalPhase = frame.kind == 'turn_completed' - ? AgentTurnPhase.finished + ? _completionPhase(frame.payload) : AgentTurnPhase.error; final turnId = frame.turnId; if (turnId != null) { @@ -204,7 +204,13 @@ List reduceAgentTurnStates( turnId: turnId, startedAt: _safeStartedAt(frame, frameAt), lastActivityAt: frameAt, - livenessTimeout: _livenessTimeout(frame.payload), + // A live-only subscription can join after turn_started. Ordinary + // ACP frames do not repeat the configured cadence, so absence here + // means unknown rather than the legacy 30-second default. + livenessTimeout: _livenessTimeout( + frame.payload, + missingFallback: _maximumTurnDuration + _livenessTimeoutSlack, + ), phase: AgentTurnPhase.working, ); } @@ -214,7 +220,12 @@ List reduceAgentTurnStates( turnsById.values.where( (turn) => !turn.isWorking || - now.difference(turn.lastActivityAt) <= turn.livenessTimeout, + (now.difference(turn.lastActivityAt) <= turn.livenessTimeout && + !now.isAfter( + turn.startedAt.add( + _maximumTurnDuration + _livenessTimeoutSlack, + ), + )), ), ); } @@ -334,9 +345,19 @@ String? _turnError(dynamic payload) { return error is String && error.trim().isNotEmpty ? error.trim() : null; } -Duration _livenessTimeout(dynamic payload) { +AgentTurnPhase _completionPhase(dynamic payload) { + if (payload is Map && payload['outcome'] == 'cancelled') { + return AgentTurnPhase.cancelled; + } + return AgentTurnPhase.finished; +} + +Duration _livenessTimeout( + dynamic payload, { + Duration missingFallback = _defaultLivenessTimeout, +}) { final rawInterval = payload is Map ? payload['livenessIntervalSecs'] : null; - if (rawInterval is! num) return _defaultLivenessTimeout; + if (rawInterval is! num) return missingFallback; final intervalSeconds = rawInterval.toInt(); if (intervalSeconds <= 0) { diff --git a/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator.dart b/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator.dart index b4c195d58..44ac2c3fd 100644 --- a/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator.dart +++ b/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator.dart @@ -875,12 +875,14 @@ class _ActivityStatusBadge extends StatelessWidget { final color = switch (status) { _ActivityStatus.working => context.appColors.success, _ActivityStatus.finished => context.colors.onSurfaceVariant, + _ActivityStatus.cancelled => context.colors.onSurfaceVariant, _ActivityStatus.error => context.colors.error, _ActivityStatus.waiting => context.appColors.warning, }; final label = switch (status) { _ActivityStatus.working => 'Working', _ActivityStatus.finished => 'Finished', + _ActivityStatus.cancelled => 'Cancelled', _ActivityStatus.error => 'Error', _ActivityStatus.waiting => 'Waiting', }; @@ -949,9 +951,12 @@ class _ActivityEmptyState extends StatelessWidget { ), const SizedBox(height: Grid.half), Text( - status == _ActivityStatus.finished - ? 'No activity rows were captured for this turn.' - : 'Waiting for live activity…', + switch (status) { + _ActivityStatus.finished => + 'No activity rows were captured for this turn.', + _ActivityStatus.cancelled => 'This turn was cancelled.', + _ => 'Waiting for live activity…', + }, style: context.textTheme.bodySmall?.copyWith( color: context.colors.onSurfaceVariant, ), diff --git a/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator/agent_activity_controls.dart b/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator/agent_activity_controls.dart index 68b380ae1..691a109ba 100644 --- a/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator/agent_activity_controls.dart +++ b/mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator/agent_activity_controls.dart @@ -380,12 +380,13 @@ class _AgentAvatarStack extends StatelessWidget { } } -enum _ActivityStatus { working, finished, error, waiting } +enum _ActivityStatus { working, finished, cancelled, error, waiting } _ActivityStatus _activityStatus(AgentTurnState? turn, bool isFallbackWorking) { return switch (turn?.phase) { AgentTurnPhase.working => _ActivityStatus.working, AgentTurnPhase.finished => _ActivityStatus.finished, + AgentTurnPhase.cancelled => _ActivityStatus.cancelled, AgentTurnPhase.error => _ActivityStatus.error, null => isFallbackWorking ? _ActivityStatus.working : _ActivityStatus.waiting, @@ -406,7 +407,10 @@ _ActivityStatus _activityStatus(AgentTurnState? turn, bool isFallbackWorking) { final errors = signals .where((signal) => signal.phase == AgentTurnPhase.error) .length; - final finished = signals.length - working - errors; + final cancelled = signals + .where((signal) => signal.phase == AgentTurnPhase.cancelled) + .length; + final finished = signals.length - working - cancelled - errors; if (working == signals.length) { return ( visibleLabel: '${signals.length} agents are working…', @@ -416,6 +420,7 @@ _ActivityStatus _activityStatus(AgentTurnState? turn, bool isFallbackWorking) { final visibleParts = [ if (working > 0) '$working working', if (finished > 0) '$finished finished', + if (cancelled > 0) '$cancelled cancelled', if (errors > 0) '$errors ${errors == 1 ? 'error' : 'errors'}', ]; final semanticParts = [ @@ -423,6 +428,8 @@ _ActivityStatus _activityStatus(AgentTurnState? turn, bool isFallbackWorking) { '$working ${working == 1 ? 'agent is' : 'agents are'} working', if (finished > 0) '$finished ${finished == 1 ? 'agent has' : 'agents have'} finished', + if (cancelled > 0) + '$cancelled ${cancelled == 1 ? 'agent was' : 'agents were'} cancelled', if (errors > 0) '$errors ${errors == 1 ? 'agent stopped' : 'agents stopped'} with ${errors == 1 ? 'an error' : 'errors'}', ]; @@ -450,6 +457,7 @@ String _selectedActivityHeadline( List transcript, ) => switch (selectedTurn?.phase) { AgentTurnPhase.finished => 'finished', + AgentTurnPhase.cancelled => 'was cancelled', AgentTurnPhase.error => 'stopped with an error', _ => transcript.isNotEmpty ? _compactHeadline(transcript.last) : 'is working…', diff --git a/mobile/test/features/channels/agent_activity/active_agent_turns_test.dart b/mobile/test/features/channels/agent_activity/active_agent_turns_test.dart index a2b6b95b2..7700fcf7d 100644 --- a/mobile/test/features/channels/agent_activity/active_agent_turns_test.dart +++ b/mobile/test/features/channels/agent_activity/active_agent_turns_test.dart @@ -52,6 +52,23 @@ void main() { expect(turns[1].errorMessage, 'Tool permission denied'); }); + test('reports cancelled completion separately from a finished turn', () { + final turns = reduceAgentTurnStates({ + 'agent-a': [ + _frame(seq: 1, second: 1, kind: 'turn_started'), + _frame( + seq: 2, + second: 2, + kind: 'turn_completed', + payload: {'outcome': 'cancelled'}, + ), + ], + }, now: DateTime.utc(2026, 8, 16, 12, 1)); + + expect(turns, hasLength(1)); + expect(turns.single.phase, AgentTurnPhase.cancelled); + }); + test('keeps an error terminal when generic completion arrives later', () { final turns = reduceAgentTurnStates({ 'agent-a': [ @@ -164,6 +181,51 @@ void main() { expect(pastBackstop, isEmpty); }); + test('does not assume a 30-second cadence when joining mid-turn', () { + final frames = { + 'agent-a': [ + _frame( + seq: 1, + second: const Duration(days: 6).inSeconds + 1, + kind: 'acp_read', + startedAt: DateTime.utc(2026, 8, 16, 12).toIso8601String(), + ), + ], + }; + + final afterLegacyTimeout = reduceAgentTurnStates( + frames, + now: DateTime.utc(2026, 8, 22, 12, 0, 32), + ); + final pastBackstop = reduceAgentTurnStates( + frames, + now: DateTime.utc(2026, 8, 23, 12, 0, 32), + ); + + expect(afterLegacyTimeout, hasLength(1)); + expect( + afterLegacyTimeout.single.livenessTimeout, + const Duration(days: 7, seconds: 30), + ); + expect(pastBackstop, isEmpty); + }); + + test('uses a liveness frame cadence when joining mid-turn', () { + final turns = reduceAgentTurnStates({ + 'agent-a': [ + _frame( + seq: 1, + second: 1, + kind: 'turn_liveness', + payload: {'livenessIntervalSecs': 120}, + ), + ], + }, now: DateTime.utc(2026, 8, 16, 12, 2, 30)); + + expect(turns, hasLength(1)); + expect(turns.single.livenessTimeout, const Duration(seconds: 150)); + }); + test('recovers a missed start and rejects stale post-terminal liveness', () { final turns = reduceAgentTurnStates({ 'agent-a': [ diff --git a/mobile/test/features/channels/agent_activity/composer_agent_activity_indicator_test.dart b/mobile/test/features/channels/agent_activity/composer_agent_activity_indicator_test.dart index e8391689b..b0d07181e 100644 --- a/mobile/test/features/channels/agent_activity/composer_agent_activity_indicator_test.dart +++ b/mobile/test/features/channels/agent_activity/composer_agent_activity_indicator_test.dart @@ -206,6 +206,53 @@ void main() { expect(find.text('Thinking'), findsOneWidget); }); + testWidgets('labels a cancelled turn separately from a finished turn', ( + tester, + ) async { + final container = ProviderContainer( + overrides: [ + composerActivityStateProvider(_scope).overrideWithValue( + const ComposerActivityState( + agents: [ + WorkingAgentSignal( + pubkey: _agentPubkey, + source: AgentWorkingSource.observer, + canViewActivity: true, + isWorking: false, + phase: AgentTurnPhase.cancelled, + turnId: _turnId, + ), + ], + humanTyping: [], + ), + ), + agentTurnStatesProvider.overrideWithValue([ + _turn(AgentTurnPhase.cancelled), + ]), + observerTurnSubscriptionProvider( + _turnKey, + ).overrideWithValue(_observerState), + userCacheProvider.overrideWith(_FakeUserCacheNotifier.new), + ], + ); + addTearDown(container.dispose); + + await tester.pumpWidget(_app(container)); + await tester.pump(); + + final control = find.byKey( + const ValueKey('composer-agent-activity-control'), + ); + expect(control, findsOneWidget); + expect(find.text('Pollen was cancelled'), findsOneWidget); + + await tester.tap(control); + await tester.pump(); + await tester.pump(const Duration(milliseconds: 240)); + + expect(find.text('Cancelled'), findsOneWidget); + }); + testWidgets('summarizes mixed working and terminal agent states', ( tester, ) async {