External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Preserve reconstructed and cancelled turn states

Browse Source 
Signed-off-by: kenny lopez <klopez4212@gmail.com>
This commit is contained in:
kenny lopez
2026-08-17 18:48:34 +01:00
parent d5b96f9a27
commit e7eb73bf53
7 changed files with 204 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-acp/src/pool.rs
+40 -7
View File
@@ -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<uuid::Uuid>,
thread_head_id: Option<String>,
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)]
docs/nips/NIP-AO.md
+10
View File
@@ -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:
mobile/lib/features/channels/agent_activity/active_agent_turns.dart
+27 -6
View File
@@ -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<AgentTurnState> 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<AgentTurnState> 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<AgentTurnState> 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) {
mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator.dart
+8 -3
View File
@@ -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,
),
mobile/lib/features/channels/agent_activity/composer_agent_activity_indicator/agent_activity_controls.dart
+10 -2
View File
@@ -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<TranscriptItem> transcript,
) => switch (selectedTurn?.phase) {
AgentTurnPhase.finished => 'finished',
AgentTurnPhase.cancelled => 'was cancelled',
AgentTurnPhase.error => 'stopped with an error',
_ =>
transcript.isNotEmpty ? _compactHeadline(transcript.last) : 'is working…',
mobile/test/features/channels/agent_activity/active_agent_turns_test.dart
+62
View File
@@ -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': [
mobile/test/features/channels/agent_activity/composer_agent_activity_indicator_test.dart
+47
View File
@@ -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 {