From fa1b27bcc8e0f076076b766273b305407b593f60 Mon Sep 17 00:00:00 2001 From: TolgaCinisli Date: Tue, 8 Sep 2026 17:05:11 +0300 Subject: [PATCH 1/2] fix(mobile): style inline code with the app mono face (#6631) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Inline code on mobile renders as **bold body text on a faint background wash** — no monospace face, no chip, and it cannot wrap. #5257 diagnosed this as a missing `highlightBuilder`. That is no longer the right fix. `gpt_markdown` 1.2.0 deprecates `highlightBuilder` (removal in 2.0.0), renders inline code as a real chip, and adds `InlineCodeStyle` for restyling it. The package author confirmed this on the issue. So this PR is an upgrade — 1.1.6 → 1.2.1 — plus one theme declaration, rather than the builder the issue originally asked for. **Where the style is declared.** `GptMarkdownThemeData` goes in `AppTheme._buildTheme`, which both `light()` and `dark()` call. That reaches all four `GptMarkdown` call sites — `message_content`, `transcript_item_widget`, `token_pill`, `custom_emoji_render` — so the style is stated once instead of per widget. A widget-level `inlineCodeStyle` would have covered channel messages only, leaving the other three on the package's defaults. **What is declared.** Face, size, ink, chip fill and outline — not the face alone. A face name on its own leaves the rest on the package's defaults, which put inline code at 14.1sp beside a fenced block's 13, on a neutral `onSurface` tint rather than the app's code surface. In dark that tint is *lighter* than the surface, while every other code surface in the app is recessed, so the chip read as a different kind of object. All of it now comes from one `CodeStyle` declaration that the fenced block reads from too, so the two cannot be edited apart. **Three adaptations the upgrade requires.** Each was found by running the gate, not by reading the changelog: 1. **`imageBuilder` widened** to `(context, url, width, height)`. This is a hard compile error, and it is **not listed in the package's migration guide**, which states "nothing here stops code compiling". Worth reporting upstream. 2. **`autolink` now defaults to `true`.** `normalizeBareLinks()` already rewrites bare URLs into Markdown links before rendering, so both would run. `message_content` opts out with `autolink: false` to keep current behaviour exactly. The migration guide argues for dropping the pre-processor instead — a better fix, but a behavioural change that belongs in its own PR. 3. **`gpt_markdown.dart` now re-exports `markdown_config.dart`**, making two direct imports redundant. `flutter analyze` reports `No issues found!` on 1.1.6 and flags both on 1.2.1, so these warnings are new, not pre-existing. **Deliberately out of scope.** The three non-message call sites now autolink bare URLs, since only `message_content` has a pre-processor to collide with. Custom inline components (`_MentionMd`, `CustomEmojiMd`, `_ChannelLinkMd`) could additionally declare `allScopesExceptLinkLabel` — 1.2.0 offers it as the fix for a `WidgetSpan` chip going blank inside a link label on iOS — but current behaviour is unchanged without it, so that stays a separate change. ### Related issue Fixes #5257 Duplicate scan: searched `gpt_markdown`, `inline code mobile`, `highlightBuilder` and `InlineCodeStyle` across both PRs and issues. No open PR touches inline code styling. #6135 (link labels) and #6166 (text selection) also touch mobile Markdown but address different defects. ### Testing Full gate, `just ci` — exit 0: | Stage | Result | |---|---| | Rust (33 suites) | 4768 passed, 0 failed | | Desktop | 5799 passed, 0 failed | | Mobile | **2011 passed**, 0 failed | | `flutter analyze` | `No issues found!` | | Desktop + web build | ok | Run on the branch with `main` merged in, so these numbers match what CI builds. **New regression test** — `renders inline code in the app code style`. It resolves the `CodeTextSpan` the package tags inline code with, which carries both the resolved `TextStyle` and the colours the chip behind it is painted with, so face, size, ink, fill and outline are all asserted rather than a widget's presence. It is negative-controlled: reverting only the theme declaration fails it with ```text Expected: a numeric value within <0.001> of <13.0> Actual: <14.1> ``` and dropping the declaration entirely falls back to `packages/gpt_markdown/JetBrainsMono` — so the test measures the real thing, and it would catch a future regression that silently drops the theme extension. The test passes `baseStyle: messageBodyTextStyle`, the style the message surfaces actually use; the widget's own fallback is the smaller `bodyMedium`, which would move the expected size. The test finds paragraphs with `find.byWidgetPredicate((widget) => widget is RichText)`, not `find.byType(RichText)`: inline code renders through `BidiRichText`, a `RichText` subclass, and `byType` matches exact runtime types. That is a hazard for any test that reads text back out of a paragraph, and one landed after this branch was cut: `message_content_custom_emoji_test.dart` arrived with #6996 and its `code keeps literal emoji while adjacent known tokens render` case reads a code span through `find.byType(RichText)`. It passes on `main` and fails on the merge result, which is what CI builds, so it went red only once CI was authorized. It now uses the same predicate. The two other `byType(RichText)` call sites — the rest of that file and `message_author_meta_test.dart` — were re-run and pass: their content carries no code span, so the exact type still matches. They were left alone. ### Screenshots Rendered through the real `MessageContent` widget with the app's own fonts loaded, at 390pt wide, 3x DPR. Sample text: ``Set `BUZZ_RELAY_URL` before launch, then run `just mobile-test` to verify.`` | | Before (1.1.6) | After (1.2.1) | |---|---|---| | Light | ![before-inline-code-light](https://raw.githubusercontent.com/TolgaCinisli/buzz/2d2d846291416d9b32d3fb9cfead950bcc4fe123/pr-6631--before-inline-code-light.png) | ![after-inline-code-light](https://raw.githubusercontent.com/TolgaCinisli/buzz/f230b95c7260a32bd5d76b1ac42130720a168521/pr-6631--after-inline-code-light.png) | | Dark | ![before-inline-code-dark](https://raw.githubusercontent.com/TolgaCinisli/buzz/2d2d846291416d9b32d3fb9cfead950bcc4fe123/pr-6631--before-inline-code-dark.png) | ![after-inline-code-dark](https://raw.githubusercontent.com/TolgaCinisli/buzz/f230b95c7260a32bd5d76b1ac42130720a168521/pr-6631--after-inline-code-dark.png) | Before: bold Inter on a flat wash, no chip edge, and `just mobile-test` breaks across the line with the wash simply ending. After: Geist Mono in a bordered, rounded chip, and the wrapped fragment gets its own chip on each line. --------- Signed-off-by: Tolga Cinisli Co-authored-by: Tolga Cinisli --- mobile/lib/app.dart | 8 +- .../transcript_item_widget.dart | 6 ++ .../features/channels/message_content.dart | 18 ++--- .../custom_emoji/custom_emoji_render.dart | 1 - .../lib/shared/theme/app_markdown_theme.dart | 46 +++++++++++ mobile/lib/shared/theme/code_style.dart | 34 +++++++++ mobile/lib/shared/theme/theme.dart | 2 + mobile/pubspec.lock | 4 +- mobile/pubspec.yaml | 2 +- .../transcript_item_widget_test.dart | 52 +++++++++++++ .../message_content_custom_emoji_test.dart | 6 +- .../channels/message_content_test.dart | 47 +++++++++++- .../shared/theme/app_markdown_theme_test.dart | 76 +++++++++++++++++++ 13 files changed, 284 insertions(+), 18 deletions(-) create mode 100644 mobile/lib/shared/theme/app_markdown_theme.dart create mode 100644 mobile/lib/shared/theme/code_style.dart create mode 100644 mobile/test/features/channels/agent_activity/transcript_item_widget_test.dart create mode 100644 mobile/test/shared/theme/app_markdown_theme_test.dart diff --git a/mobile/lib/app.dart b/mobile/lib/app.dart index d87982bac23..9f3724fd2d1 100644 --- a/mobile/lib/app.dart +++ b/mobile/lib/app.dart @@ -374,9 +374,11 @@ class App extends HookConsumerWidget { // Above the navigator, so a burst keeps playing over a pushed thread page // or a modal sheet — the same reason desktop pins its canvas to the // viewport rather than to the message row. - builder: (context, child) => MobileHuddleShell( - navigatorKey: _mobileRootNavigatorKey, - child: EmojiBurstOverlay(child: child ?? const SizedBox.shrink()), + builder: (context, child) => AppMarkdownTheme( + child: MobileHuddleShell( + navigatorKey: _mobileRootNavigatorKey, + child: EmojiBurstOverlay(child: child ?? const SizedBox.shrink()), + ), ), home: authState.when( loading: () => const _SplashScreen(), diff --git a/mobile/lib/features/channels/agent_activity/transcript_item_widget.dart b/mobile/lib/features/channels/agent_activity/transcript_item_widget.dart index 74b2322e34a..b2f515a888a 100644 --- a/mobile/lib/features/channels/agent_activity/transcript_item_widget.dart +++ b/mobile/lib/features/channels/agent_activity/transcript_item_widget.dart @@ -72,6 +72,10 @@ class _MessageItemWidget extends StatelessWidget { style: context.textTheme.bodyMedium?.copyWith( color: context.colors.onSurface, ), + // No link handler here, so an autolinked URL would draw as a + // link and do nothing when tapped. Transcript URLs stay text, + // as they were before gpt_markdown started autolinking. + autolink: false, ), ], ), @@ -143,6 +147,8 @@ class _ThoughtItemWidget extends HookWidget { style: context.textTheme.bodySmall?.copyWith( color: context.colors.onSurfaceVariant, ), + // As above: no link handler on this surface either. + autolink: false, ), ], ], diff --git a/mobile/lib/features/channels/message_content.dart b/mobile/lib/features/channels/message_content.dart index c9aa72599fb..b94954c00cf 100644 --- a/mobile/lib/features/channels/message_content.dart +++ b/mobile/lib/features/channels/message_content.dart @@ -7,7 +7,6 @@ import 'package:flutter/services.dart'; import 'package:flutter_hooks/flutter_hooks.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:gpt_markdown/gpt_markdown.dart'; -import 'package:gpt_markdown/custom_widgets/markdown_config.dart'; import 'package:lucide_icons_flutter/lucide_icons.dart'; import 'package:open_filex/open_filex.dart'; import 'package:path_provider/path_provider.dart'; @@ -266,6 +265,9 @@ class MessageContent extends HookConsumerWidget { finalContent, style: style, followLinkColor: false, + // normalizeBareLinks() already turns bare URLs into Markdown links; + // gpt_markdown 1.2.0 autolinks by default, so both would run. + autolink: false, codeBuilder: (context, name, code, closed) => _MessageCodeBlock(name: name, code: code), linkBuilder: (context, linkText, url, linkStyle) => _buildLink( @@ -279,7 +281,7 @@ class MessageContent extends HookConsumerWidget { resolvedChannelTap, resolvedChannelNames, ), - imageBuilder: (context, imageUrl) => + imageBuilder: (context, imageUrl, _, _) => _buildMedia(context, imageUrl, imetaByUrl[imageUrl]), textAlign: textAlign, maxLines: maxLines, @@ -721,9 +723,9 @@ class _MessageCodeBlock extends HookWidget { } final codeBaseStyle = TextStyle( - fontFamily: 'GeistMono', - fontSize: 13, - height: 1.5, + fontFamily: CodeStyle.fontFamily, + fontSize: CodeStyle.fontSize, + height: CodeStyle.lineHeight, color: context.colors.onSurface, ); final isDark = context.theme.brightness == Brightness.dark; @@ -735,11 +737,9 @@ class _MessageCodeBlock extends HookWidget { return Container( margin: const EdgeInsets.only(top: Grid.half), decoration: BoxDecoration( - color: context.colors.surfaceContainerHighest.withValues(alpha: 0.6), + color: CodeStyle.background(context.colors), borderRadius: BorderRadius.circular(12), - border: Border.all( - color: context.colors.outline.withValues(alpha: 0.7), - ), + border: Border.all(color: CodeStyle.border(context.colors)), ), child: Column( crossAxisAlignment: CrossAxisAlignment.stretch, diff --git a/mobile/lib/shared/custom_emoji/custom_emoji_render.dart b/mobile/lib/shared/custom_emoji/custom_emoji_render.dart index feede9aec81..bbfdd25603e 100644 --- a/mobile/lib/shared/custom_emoji/custom_emoji_render.dart +++ b/mobile/lib/shared/custom_emoji/custom_emoji_render.dart @@ -1,6 +1,5 @@ import 'package:flutter/material.dart'; import 'package:gpt_markdown/gpt_markdown.dart'; -import 'package:gpt_markdown/custom_widgets/markdown_config.dart'; import '../relay/relay.dart'; diff --git a/mobile/lib/shared/theme/app_markdown_theme.dart b/mobile/lib/shared/theme/app_markdown_theme.dart new file mode 100644 index 00000000000..27e5060901d --- /dev/null +++ b/mobile/lib/shared/theme/app_markdown_theme.dart @@ -0,0 +1,46 @@ +import 'package:flutter/material.dart'; +import 'package:gpt_markdown/gpt_markdown.dart'; + +import 'app_theme.dart'; +import 'code_style.dart'; + +/// States the app's code style on every `GptMarkdown` below it and leaves the +/// rest of the Markdown theme as it already was. +/// +/// Registering a [GptMarkdownThemeData] as a `ThemeExtension` would be the +/// shorter route, but its factory is not a partial override: it builds a fresh +/// stock Material theme and fills every heading, rule and highlight field from +/// it, so registering one stops `GptMarkdownTheme.of` deriving those from the +/// app. Restating them in the extension does not close the gap either — the +/// ambient headings carry the localized text geometry, which exists only once +/// `Theme.of` has run and a `ThemeData` factory cannot see. Overriding the one +/// field on the ambient theme is what leaves non-code Markdown untouched. +class AppMarkdownTheme extends StatelessWidget { + const AppMarkdownTheme({super.key, required this.child}); + + final Widget child; + + @override + Widget build(BuildContext context) { + final scheme = Theme.of(context).colorScheme; + return GptMarkdownTheme( + gptThemeData: GptMarkdownTheme.of(context).copyWith( + // Inline code is otherwise drawn in gpt_markdown's own bundled face, at + // its own size, tinted from `onSurface`. Stating the app's code style + // gives every Markdown surface the face, size and chip colours a fenced + // code block already uses. + inlineCode: InlineCodeStyle( + fontFamily: CodeStyle.fontFamily, + fontSizeFactor: CodeStyle.fontSizeFactor, + color: scheme.onSurface, + backgroundColor: CodeStyle.background(scheme), + borderColor: CodeStyle.border(scheme), + // A chip is small, so it takes the smallest step of the app's radius + // scale rather than the `Radii.card` a block uses. + borderRadius: const Radius.circular(Radii.xs), + ).resolve(scheme), + ), + child: child, + ); + } +} diff --git a/mobile/lib/shared/theme/code_style.dart b/mobile/lib/shared/theme/code_style.dart new file mode 100644 index 00000000000..3ed667f44c6 --- /dev/null +++ b/mobile/lib/shared/theme/code_style.dart @@ -0,0 +1,34 @@ +import 'package:flutter/material.dart'; + +import 'message_typography.dart'; + +/// The app's own code style — one place stating the mono face, its size and +/// the colours a code surface is drawn with. +/// +/// Two surfaces render code and they have to read as one family: the fenced +/// block in `MessageContent`, and the inline chip `gpt_markdown` paints from +/// the `InlineCodeStyle` declared in `AppTheme`. Both take their values from +/// here, so neither can drift from the other by editing one copy. +abstract final class CodeStyle { + /// The mono face, already used by the composer and by code blocks. + static const fontFamily = 'GeistMono'; + + /// Code size at message body scale. + static const fontSize = 13.0; + + /// Line height of code text. + static const lineHeight = 1.5; + + /// [fontSize] stated against the message body size, for APIs that scale code + /// relative to the text around it rather than fixing a point size. Keeping + /// it a ratio means code follows the body if that size ever changes. + static final fontSizeFactor = fontSize / messageBodyTextStyle.fontSize!; + + /// Fill behind a code surface. + static Color background(ColorScheme scheme) => + scheme.surfaceContainerHighest.withValues(alpha: 0.6); + + /// Outline around a code surface. + static Color border(ColorScheme scheme) => + scheme.outline.withValues(alpha: 0.7); +} diff --git a/mobile/lib/shared/theme/theme.dart b/mobile/lib/shared/theme/theme.dart index eacd1ea7807..55615b5b734 100644 --- a/mobile/lib/shared/theme/theme.dart +++ b/mobile/lib/shared/theme/theme.dart @@ -1,8 +1,10 @@ export 'accent_colors.dart'; export 'adaptive_theme.dart'; export 'app_colors.dart'; +export 'app_markdown_theme.dart'; export 'app_theme.dart'; export 'buzz_theme.dart'; +export 'code_style.dart'; export 'color_scheme.dart'; export 'community_theme_preference.dart'; export 'community_theme_provider.dart'; diff --git a/mobile/pubspec.lock b/mobile/pubspec.lock index 8b190dcb03c..0632d4a8df7 100644 --- a/mobile/pubspec.lock +++ b/mobile/pubspec.lock @@ -620,10 +620,10 @@ packages: dependency: "direct main" description: name: gpt_markdown - sha256: "5c565a438e569b3b546023d0d4eccccf87e260a3289a17c4e241c41d5e7545df" + sha256: "42887afaf5c7a71348221048a6c1e34c9836703ee0e6b4865e789253fcaa204a" url: "https://pub.dev" source: hosted - version: "1.1.6" + version: "1.2.1" gtk: dependency: transitive description: diff --git a/mobile/pubspec.yaml b/mobile/pubspec.yaml index a5166bfa075..e6806b29d19 100644 --- a/mobile/pubspec.yaml +++ b/mobile/pubspec.yaml @@ -27,7 +27,7 @@ dependencies: nostr: ^2.0.0 pointycastle: ^4.0.0 url_launcher: ^6.3.2 - gpt_markdown: ^1.1.6 + gpt_markdown: ^1.2.1 highlight: ^0.7.0 intl: ^0.20.2 uuid: ^4.5.1 diff --git a/mobile/test/features/channels/agent_activity/transcript_item_widget_test.dart b/mobile/test/features/channels/agent_activity/transcript_item_widget_test.dart new file mode 100644 index 00000000000..20524e9a2db --- /dev/null +++ b/mobile/test/features/channels/agent_activity/transcript_item_widget_test.dart @@ -0,0 +1,52 @@ +import 'package:flutter/material.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:gpt_markdown/custom_widgets/link_button.dart'; +import 'package:buzz/features/channels/agent_activity/observer_models.dart'; +import 'package:buzz/features/channels/agent_activity/transcript_item_widget.dart'; +import 'package:buzz/shared/theme/theme.dart'; + +Widget _testable(Widget child) { + return MaterialApp( + theme: AppTheme.light(), + home: Scaffold(body: SingleChildScrollView(child: child)), + ); +} + +void main() { + // Neither transcript surface passes a link handler, so an autolinked bare + // URL would draw as a link and then do nothing when tapped. They stay text. + testWidgets('leaves a bare URL in a message as plain text', (tester) async { + await tester.pumpWidget( + _testable( + TranscriptItemWidget( + item: MessageItem( + id: 'm1', + role: 'assistant', + title: 'Assistant', + text: 'Report is at https://example.com/report today.', + timestamp: '2026-08-29T09:00:00Z', + ), + ), + ), + ); + + expect(find.byType(LinkButton), findsNothing); + }); + + testWidgets('leaves a bare URL in a thought as plain text', (tester) async { + await tester.pumpWidget( + _testable( + TranscriptItemWidget( + item: ThoughtItem( + id: 't1', + title: 'Thinking', + text: 'Checking https://example.com/report first.', + timestamp: '2026-08-29T09:00:00Z', + ), + ), + ), + ); + + expect(find.byType(LinkButton), findsNothing); + }); +} diff --git a/mobile/test/features/channels/message_content_custom_emoji_test.dart b/mobile/test/features/channels/message_content_custom_emoji_test.dart index 3059af9a93e..c43ed614c68 100644 --- a/mobile/test/features/channels/message_content_custom_emoji_test.dart +++ b/mobile/test/features/channels/message_content_custom_emoji_test.dart @@ -111,8 +111,12 @@ void main() { ) async { await tester.pumpWidget(_testable('`:wave:` :unknown:wave:')); expect(find.byType(CustomEmojiImage), findsOneWidget); + // A paragraph holding inline code renders through BidiRichText, a RichText + // subclass, and `find.byType` matches the exact runtime type only. final text = tester - .widgetList(find.byType(RichText)) + .widgetList( + find.byWidgetPredicate((widget) => widget is RichText), + ) .map((widget) => widget.text.toPlainText()) .join(); expect(text, contains(':wave:')); diff --git a/mobile/test/features/channels/message_content_test.dart b/mobile/test/features/channels/message_content_test.dart index aaf6a5973b1..8fe47e88030 100644 --- a/mobile/test/features/channels/message_content_test.dart +++ b/mobile/test/features/channels/message_content_test.dart @@ -45,7 +45,8 @@ Widget _testable( data: MediaQuery.of( context, ).copyWith(disableAnimations: disableAnimations), - child: Scaffold(body: child), + // The app states its code style here, above the navigator. + child: AppMarkdownTheme(child: Scaffold(body: child)), ), ), ), @@ -2813,6 +2814,50 @@ Photos expect(_allRichText(tester), isNot(contains('**'))); }); + testWidgets('renders inline code in the app code style', (tester) async { + // The message surfaces pass this style in; the widget's own fallback + // is the smaller `bodyMedium`, which would move the expected size. + await tester.pumpWidget( + _testable( + const MessageContent( + content: 'Run `just test` now', + baseStyle: messageBodyTextStyle, + ), + ), + ); + + // gpt_markdown tags inline code with a CodeTextSpan carrying both the + // resolved text style and the colours the chip behind it is painted + // with. It renders through BidiRichText, a RichText subclass, so + // find.byType(RichText) would miss it. + final codeSpans = []; + for (final rich in tester.widgetList( + find.byWidgetPredicate((widget) => widget is RichText), + )) { + rich.text.visitChildren((span) { + if (span is CodeTextSpan && span.text == 'just test') { + codeSpans.add(span); + } + return true; + }); + } + + expect(codeSpans, hasLength(1)); + final code = codeSpans.single; + const scheme = lightColorScheme; + + // Face, size and ink — the values a fenced code block is drawn with, + // not the package's bundled mono at its own 0.94 of the body size. + expect(code.style?.fontFamily, CodeStyle.fontFamily); + expect(code.style?.fontSize, closeTo(CodeStyle.fontSize, 0.001)); + expect(code.style?.color, scheme.onSurface); + + // Chip fill and outline come from the app's code surface rather than + // the package's `onSurface` tints. + expect(code.codeStyle.backgroundColor, CodeStyle.background(scheme)); + expect(code.codeStyle.borderColor, CodeStyle.border(scheme)); + }); + testWidgets('renders code block between paragraphs', (tester) async { await tester.pumpWidget( _testable( diff --git a/mobile/test/shared/theme/app_markdown_theme_test.dart b/mobile/test/shared/theme/app_markdown_theme_test.dart new file mode 100644 index 00000000000..26ae369821a --- /dev/null +++ b/mobile/test/shared/theme/app_markdown_theme_test.dart @@ -0,0 +1,76 @@ +import 'package:flutter/material.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:gpt_markdown/gpt_markdown.dart'; +import 'package:buzz/shared/theme/app_markdown_theme.dart'; +import 'package:buzz/shared/theme/app_theme.dart'; +import 'package:buzz/shared/theme/code_style.dart'; + +/// The Markdown theme a widget reads under [theme], with [wrapped] deciding +/// whether the app states its code style above it. +Future _markdownTheme( + WidgetTester tester, + ThemeData theme, { + required bool wrapped, +}) async { + late GptMarkdownThemeData read; + final reader = Builder( + builder: (context) { + read = GptMarkdownTheme.of(context); + return const SizedBox.shrink(); + }, + ); + + await tester.pumpWidget( + MaterialApp( + theme: theme, + home: wrapped ? AppMarkdownTheme(child: reader) : reader, + ), + ); + // MaterialApp animates a theme change, so the first frame after a pump can + // still carry the previous one. + await tester.pumpAndSettle(); + return read; +} + +void main() { + testWidgets('states inline code and leaves the rest of Markdown alone', ( + tester, + ) async { + for (final theme in [AppTheme.light(), AppTheme.dark()]) { + final ambient = await _markdownTheme(tester, theme, wrapped: false); + final stated = await _markdownTheme(tester, theme, wrapped: true); + + // Headings, rules and the rest keep the values gpt_markdown derives from + // the app theme: a code-only change must not restyle anything else. + expect(stated.h1, ambient.h1); + expect(stated.h2, ambient.h2); + expect(stated.h3, ambient.h3); + expect(stated.h4, ambient.h4); + expect(stated.h5, ambient.h5); + expect(stated.h6, ambient.h6); + expect(stated.hrLineColor, ambient.hrLineColor); + expect(stated.hrLineThickness, ambient.hrLineThickness); + expect(stated.hrLinePadding, ambient.hrLinePadding); + expect(stated.highlightColor, ambient.highlightColor); + expect(stated.linkColor, ambient.linkColor); + expect(stated.linkHoverColor, ambient.linkHoverColor); + expect( + stated.autoAddDividerLineAfterH1, + ambient.autoAddDividerLineAfterH1, + ); + expect(stated.styleSheet, ambient.styleSheet); + + // Inline code is the whole of the difference. + final scheme = theme.colorScheme; + expect(stated.inlineCode.fontFamily, CodeStyle.fontFamily); + expect(stated.inlineCode.color, scheme.onSurface); + expect(stated.inlineCode.backgroundColor, CodeStyle.background(scheme)); + expect(stated.inlineCode.borderColor, CodeStyle.border(scheme)); + expect( + ambient.inlineCode.fontFamily, + isNot(CodeStyle.fontFamily), + reason: 'the package default is what this widget exists to replace', + ); + } + }); +} From 86c189e8571d4254726f4d0519a500df9b9d947e Mon Sep 17 00:00:00 2001 From: Salman Mohammed Date: Tue, 8 Sep 2026 10:10:38 -0400 Subject: [PATCH 2/2] fix(buzz-acp): wake held ACP threads and fence forked sessions (#7340) ## Summary Adds an independent deadline wakeup so held thread work dispatches after its 10-second bound even when the relay loop is otherwise quiet. Fences session ownership by generation so a worker returning after a fork cannot make an older provider session claimable again. This follows up on the two post-merge findings from [#7337](https://github.com/block/buzz/pull/7337#pullrequestreview-5116329341). ### Related issue Follow-up to #7337. ### Testing - `cargo test -p buzz-acp` - `cargo clippy -p buzz-acp --all-targets -- -D warnings` - Pre-push file-size, differential Rust test, and desktop Tauri gates No UI changes. --- **Update Sep 4, 15:35:** Addressed both Codex review findings. - Queue-cap eviction now prunes orphaned hold deadlines. - An expired hold stays expired until a worker is successfully claimed. - Hold timers remain disabled while every worker is busy; worker return wakes dispatch directly. - Added regressions for queue eviction and pool exhaustion. Generated with Codex --------- Signed-off-by: Salman Mohammed --- crates/buzz-acp/src/lib.rs | 167 +++++++++++++---- crates/buzz-acp/src/pool.rs | 344 ++++++++++++++++++++++++++++++++--- crates/buzz-acp/src/queue.rs | 16 ++ 3 files changed, 466 insertions(+), 61 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index ddd594b142d..cc2952a2f8d 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -3005,6 +3005,7 @@ async fn tokio_main() -> Result<()> { Panic(tokio::task::JoinError), SteerAck(SteerAckEvent), Wake(u32, Result), + HoldDeadline, } loop { @@ -3146,6 +3147,8 @@ async fn tokio_main() -> Result<()> { } // Borrow result_rx and join_set simultaneously via split-borrow helper. + pool.retain_held_scopes(|scope| queue.has_pending_scope(scope)); + let hold_deadline = pool.next_hold_deadline(pool::HOLD_BUSY_OWNER_TIMEOUT); let pool_event: Option = { let (result_rx, join_set) = pool.rx_and_join_set(); tokio::select! { @@ -3188,6 +3191,9 @@ async fn tokio_main() -> Result<()> { _ => std::future::pending().await, } } => None, + _ = pool::AgentPool::wait_for_hold_deadline(hold_deadline), if pool_ready => { + Some(PoolEvent::HoldDeadline) + }, Some(Err(error)) = wake_tasks.join_next(), if !wake_tasks.is_empty() => { if let Some(attempt) = pool_lifecycle.waking_attempt() { let message = format!("pool wake task failed: {error}"); @@ -4021,6 +4027,21 @@ async fn tokio_main() -> Result<()> { } } } + Some(PoolEvent::HoldDeadline) => { + // A held thread must make progress even when every unrelated + // relay/timer source is quiet. The deadline is derived from + // the pool's first-held stamp, so this dispatch observes + // `ForkAfterHold` and claims an idle worker immediately. + for (scope, thread_tags) in dispatch_pending( + &mut pool, + &mut queue, + &ctx, + &mut last_activity, + observer.as_ref(), + ) { + typing_channels.insert(scope, thread_tags); + } + } None => {} // relay/heartbeat/shutdown branches handled inline above } } @@ -4412,7 +4433,7 @@ fn dispatch_pending( let mut held: Vec = Vec::new(); // One clock read for the whole cycle so every batch's bounded-hold window is // measured against the same instant. - let now = std::time::Instant::now(); + let now = tokio::time::Instant::now(); loop { let batch = match queue.flush_next() { Some(b) => b, @@ -4427,7 +4448,8 @@ fn dispatch_pending( // so an active channel cannot starve a sibling channel on a shared // worker. A held thread that outwaits the window forks a fresh session // rather than starve behind an unbounded turn. - match pool.hold_decision(&scope, now, pool::HOLD_BUSY_OWNER_TIMEOUT) { + let forked_after_hold = match pool.hold_decision(&scope, now, pool::HOLD_BUSY_OWNER_TIMEOUT) + { pool::HoldDecision::Hold { held_for, owner_index, @@ -4458,31 +4480,9 @@ fn dispatch_pending( pool::HoldDecision::ForkAfterHold { held_for, owner_index, - } => { - tracing::warn!( - channel = %channel_id, - scope = %scope.telemetry_label(), - owner_index, - held_for_secs = held_for.as_secs_f64(), - "busy-owner hold expired — forking fresh session on an idle worker" - ); - if let Some(observer) = observer { - observer.emit( - "busy_owner_hold_forked", - None, - &observer::context_for(Some(channel_id), None, None), - serde_json::json!({ - "scope": scope.telemetry_label(), - "ownerIndex": owner_index, - "heldForSecs": held_for.as_secs_f64(), - }), - ); - } - // Fall through to try_claim below (fork); record_scope_owner - // reassigns ownership to the new worker automatically. - } - pool::HoldDecision::Dispatch => {} - } + } => Some((held_for, owner_index)), + pool::HoldDecision::Dispatch => None, + }; let typing_scope = batch .events .last() @@ -4502,6 +4502,32 @@ fn dispatch_pending( break; } }; + // Consume a bounded hold only after a worker was actually claimed. + // If every slot is checked out, the expired stamp remains sticky and + // the worker-return event retries immediately instead of waiting for a + // fresh timeout window. + pool.clear_hold(&scope); + if let Some((held_for, owner_index)) = forked_after_hold { + tracing::warn!( + channel = %channel_id, + scope = %scope.telemetry_label(), + owner_index, + held_for_secs = held_for.as_secs_f64(), + "busy-owner hold expired — forking fresh session on an idle worker" + ); + if let Some(observer) = observer { + observer.emit( + "busy_owner_hold_forked", + None, + &observer::context_for(Some(channel_id), None, None), + serde_json::json!({ + "scope": scope.telemetry_label(), + "ownerIndex": owner_index, + "heldForSecs": held_for.as_secs_f64(), + }), + ); + } + } tracing::debug!(agent = agent.index, channel = %channel_id, scope = %scope.telemetry_label(), affinity_hit, "agent_claimed"); let recoverable_batch = match ctx.dedup_mode { @@ -4532,6 +4558,14 @@ fn dispatch_pending( let turn_id = Uuid::new_v4().to_string(); let task_turn_id = turn_id.clone(); + // Assign ownership before moving the worker into the task. If this is + // a bounded-hold fork, the new generation immediately invalidates the + // prior busy worker's copy when that worker eventually returns. + let owner_generation = pool.record_scope_owner(scope.clone(), agent.index); + agent + .state + .set_scope_owner_generation(scope.clone(), owner_generation); + let abort_handle = pool.join_set.spawn(async move { pool::run_prompt_task( agent, @@ -4558,9 +4592,6 @@ fn dispatch_pending( successful_steer_deliveries: HashSet::new(), }, ); - // Record this worker as the scope's session owner so a later dispatch - // while it is busy holds instead of forking a duplicate session. - pool.record_scope_owner(scope.clone(), agent_index); dispatched_channels.push((scope, typing_scope)); *last_activity = tokio::time::Instant::now(); } @@ -6235,7 +6266,7 @@ mod owner_control_command_tests { // The bounded hold decision stamps A's first-held time, then forks once // the window elapses rather than starving behind the busy owner. - let now = std::time::Instant::now(); + let now = tokio::time::Instant::now(); assert!( matches!( pool.hold_decision(&ta, now, pool::HOLD_BUSY_OWNER_TIMEOUT), @@ -6256,9 +6287,10 @@ mod owner_control_command_tests { "elapsed window => fork on an idle worker" ); assert!( - !pool.held_since_contains(&ta), - "fork clears the first-held stamp" + pool.held_since_contains(&ta), + "expired hold stays sticky until an idle worker is claimed" ); + pool.clear_hold(&ta); // A conversation scope never holds even with a busy recorded owner — // this is the cross-channel head-of-line-blocking regression guard. @@ -6290,6 +6322,55 @@ mod owner_control_command_tests { ); } + #[tokio::test] + async fn queue_cap_eviction_prunes_orphaned_hold_deadline() { + let mut pool = AgentPool::from_slots(vec![]); + let mut queue = EventQueue::new(DedupMode::Queue); + let channel_id = Uuid::new_v4(); + let held_scope = thread_scope(channel_id, &"a".repeat(64)); + let surviving_scope = thread_scope(channel_id, &"b".repeat(64)); + + pool.record_scope_owner(held_scope.clone(), 0); + let (tx, _rx) = tokio::sync::oneshot::channel(); + insert_task_meta(&mut pool, 0, surviving_scope.clone(), tx); + assert!(matches!( + pool.hold_decision( + &held_scope, + tokio::time::Instant::now(), + pool::HOLD_BUSY_OWNER_TIMEOUT + ), + pool::HoldDecision::Hold { .. } + )); + + let oldest = std::time::Instant::now() - Duration::from_secs(1); + queue.push(queue::QueuedEvent { + channel_id, + scope: held_scope.clone(), + event: make_event(KIND_STREAM_MESSAGE, "held", None), + received_at: oldest, + prompt_tag: "test".into(), + }); + for i in 0..500 { + queue.push(queue::QueuedEvent { + channel_id, + scope: surviving_scope.clone(), + event: make_event(KIND_STREAM_MESSAGE, &format!("new-{i}"), None), + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + }); + } + + assert!( + !queue.has_pending_scope(&held_scope), + "aggregate cap evicts the globally oldest scope" + ); + pool.retain_held_scopes(|scope| queue.has_pending_scope(scope)); + assert!( + !pool.held_since_contains(&held_scope), + "evicted scope cannot leave an immediately-ready deadline behind" + ); + } + #[test] fn project_owner_control_signs_only_addressable_project_events() { let keys = Keys::generate(); @@ -9370,6 +9451,15 @@ mod error_outcome_emission_tests { } } + fn bind_agent_scope_owner( + pool: &mut AgentPool, + agent: &mut OwnedAgent, + scope: scope::SessionScope, + ) { + let generation = pool.record_scope_owner(scope.clone(), agent.index); + agent.state.set_scope_owner_generation(scope, generation); + } + #[tokio::test] async fn successful_native_steer_is_transferred_to_live_session_delivery_state() { let channel_id = Uuid::new_v4(); @@ -9385,6 +9475,11 @@ mod error_outcome_emission_tests { ); let mut pool = AgentPool::from_slots(vec![None]); + bind_agent_scope_owner( + &mut pool, + &mut agent, + scope::SessionScope::Conversation { channel_id }, + ); let task_id = pool.join_set.spawn(async {}).id(); pool.task_map_mut().insert( task_id, @@ -9460,6 +9555,11 @@ mod error_outcome_emission_tests { ); let mut pool = AgentPool::from_slots(vec![None]); + bind_agent_scope_owner( + &mut pool, + &mut agent, + scope::SessionScope::Conversation { channel_id }, + ); let task_id = pool.join_set.spawn(async {}).id(); pool.task_map_mut().insert( task_id, @@ -10697,6 +10797,7 @@ mod error_outcome_emission_tests { .sessions .insert(session_scope.clone(), "healthy-session".into()); let mut pool = AgentPool::from_slots(vec![None]); + bind_agent_scope_owner(&mut pool, &mut agent, session_scope.clone()); let task_id = pool.join_set.spawn(async {}).id(); pool.task_map_mut().insert( task_id, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 06383d456d3..dbeafedda70 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -141,9 +141,18 @@ pub struct SessionState { /// Per-scope successful-delivery state. Created with the ACP session and /// cleared atomically with every invalidation path. pub deliveries: HashMap, + /// Pool-assigned ownership generation for each scope. A worker returning + /// after another worker forked the scope carries an older generation; the + /// pool uses this fence to discard that stale provider session before the + /// worker becomes claimable again. + scope_owner_generations: HashMap, } impl SessionState { + pub(crate) fn set_scope_owner_generation(&mut self, scope: SessionScope, generation: u64) { + self.scope_owner_generations.insert(scope, generation); + } + /// Invalidate the session (and turn counter) for a specific prompt source. pub fn invalidate(&mut self, source: &PromptSource) { match source { @@ -165,6 +174,7 @@ impl SessionState { self.core_sections.remove(scope); self.canvas_sections.remove(scope); self.deliveries.remove(scope); + self.scope_owner_generations.remove(scope); self.sessions.remove(scope).is_some() } @@ -179,6 +189,7 @@ impl SessionState { .chain(self.core_sections.keys()) .chain(self.canvas_sections.keys()) .chain(self.deliveries.keys()) + .chain(self.scope_owner_generations.keys()) .filter(|s| s.channel_id() == *channel_id) .cloned() .collect::>() @@ -203,6 +214,7 @@ impl SessionState { self.core_sections.clear(); self.canvas_sections.clear(); self.deliveries.clear(); + self.scope_owner_generations.clear(); } pub(crate) fn mark_scope_delivery_success( @@ -336,13 +348,23 @@ pub struct AgentPool { /// cause another worker to open a duplicate session for the same thread. /// Best-effort: stale entries (rotation, crash/respawn) self-heal on the /// next dispatch and are pruned on channel-wide session invalidation. - session_owners: HashMap, + session_owners: HashMap, + /// Monotonic validity fence assigned whenever a scope is dispatched. The + /// generation distinguishes a newly forked owner from every older copy of + /// that scope's provider session. + next_scope_owner_generation: u64, /// First time each scope was held for a busy owner, so the bounded hold can /// expire and fork rather than starve behind an unbounded turn. Derived /// state: cleared on every dispatch/invalidation path, and only ever holds /// `Thread` scopes (the sole variant [`hold_decision`](Self::hold_decision) /// stamps). - held_since: HashMap, + held_since: HashMap, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct SessionOwner { + agent_index: usize, + generation: u64, } /// Result returned by a completed prompt task. @@ -830,14 +852,33 @@ impl AgentPool { join_set: JoinSet::new(), task_map: HashMap::new(), session_owners: HashMap::new(), + next_scope_owner_generation: 1, held_since: HashMap::new(), } } - /// Record which worker is handling `scope` so a later dispatch can detect a - /// busy owner and avoid opening a duplicate session on another worker. - pub fn record_scope_owner(&mut self, scope: SessionScope, agent_index: usize) { - self.session_owners.insert(scope, agent_index); + /// Record `agent_index` as the newest owner of `scope`, returning the + /// generation that the caller must install on the checked-out worker. + /// Returning workers are accepted only while this exact + /// `(worker, generation)` pair remains authoritative. + pub fn record_scope_owner(&mut self, scope: SessionScope, agent_index: usize) -> u64 { + let generation = self.next_scope_owner_generation; + self.next_scope_owner_generation = self.next_scope_owner_generation.wrapping_add(1); + if self.next_scope_owner_generation == 0 { + // Preserve zero as an unassigned sentinel. Reaching this requires + // 2^64 dispatches in one process, but resetting safely is cheap: + // every previously tagged session becomes stale on return/claim. + self.next_scope_owner_generation = 1; + self.session_owners.clear(); + } + self.session_owners.insert( + scope, + SessionOwner { + agent_index, + generation, + }, + ); + generation } /// True when this scope should be **held** (left queued) rather than @@ -854,16 +895,21 @@ impl AgentPool { return false; } match self.session_owners.get(scope) { - Some(&owner_idx) => self.task_map.values().any(|m| m.agent_index == owner_idx), + Some(owner) => self + .task_map + .values() + .any(|m| m.agent_index == owner.agent_index), None => false, } } /// Decide whether to hold `scope`'s batch for its busy session owner, fork it - /// after a bounded hold, or dispatch immediately. Stamps and clears the - /// first-held time internally so the bounded window survives across dispatch - /// cycles without a dedicated timer; `now` and `timeout` are injected for - /// testability. + /// after a bounded hold, or dispatch immediately. Stamps the first-held time + /// so the bounded window survives across dispatch cycles; `now` and + /// `timeout` are injected for testability. An expired + /// stamp remains sticky until [`clear_hold`](Self::clear_hold) confirms a + /// worker was successfully claimed, so pool exhaustion cannot restart the + /// bounded window. /// /// Gated on the scope variant, not the session policy: `Conversation` scopes /// (channel-policy channels and all DMs) never hold — a busy owner there means @@ -873,18 +919,21 @@ impl AgentPool { pub fn hold_decision( &mut self, scope: &SessionScope, - now: std::time::Instant, + now: tokio::time::Instant, timeout: Duration, ) -> HoldDecision { if !scope.is_thread() || !self.should_hold_for_busy_owner(scope) { self.held_since.remove(scope); return HoldDecision::Dispatch; } - let owner_index = self.session_owners.get(scope).copied().unwrap_or_default(); + let owner_index = self + .session_owners + .get(scope) + .map(|owner| owner.agent_index) + .unwrap_or_default(); let first = *self.held_since.entry(scope.clone()).or_insert(now); let held_for = now.saturating_duration_since(first); if held_for >= timeout { - self.held_since.remove(scope); HoldDecision::ForkAfterHold { held_for, owner_index, @@ -911,7 +960,7 @@ impl AgentPool { if let Some(scope) = scope { let idx = self.agents.iter().position(|slot| { slot.as_ref() - .map(|a| a.state.sessions.contains_key(scope)) + .map(|a| self.agent_owns_scope(a, scope)) .unwrap_or(false) }); if let Some(i) = idx { @@ -925,7 +974,27 @@ impl AgentPool { } /// Return an agent to its slot after a task completes. - pub fn return_agent(&mut self, agent: OwnedAgent) { + pub fn return_agent(&mut self, mut agent: OwnedAgent) { + let stale_scopes: Vec = agent + .state + .sessions + .keys() + .filter(|scope| !self.agent_owns_scope(&agent, scope)) + .cloned() + .collect(); + for scope in stale_scopes { + tracing::info!( + agent = agent.index, + scope = %scope.telemetry_label(), + "discarding stale session after ownership changed" + ); + agent.state.invalidate_scope(&scope); + } + let live_scopes: HashSet = agent.state.sessions.keys().cloned().collect(); + agent + .state + .scope_owner_generations + .retain(|scope, _| live_scopes.contains(scope)); let idx = agent.index; if self.agents[idx].is_some() { // This is a bug: two tasks returned the same agent index. Log it @@ -945,16 +1014,64 @@ impl AgentPool { self.agents.iter().any(|slot| slot.is_some()) } + /// Confirm that pending work for `scope` successfully claimed a worker. + /// + /// In particular, an expired busy-owner hold must not be consumed until + /// this point: `try_claim` can fail while every worker remains checked out. + pub(crate) fn clear_hold(&mut self, scope: &SessionScope) { + self.held_since.remove(scope); + } + + /// Remove derived hold stamps for scopes that no longer have pending work. + pub(crate) fn retain_held_scopes( + &mut self, + mut has_pending_work: impl FnMut(&SessionScope) -> bool, + ) { + self.held_since.retain(|scope, _| has_pending_work(scope)); + } + /// Whether any idle agent already has a session for `scope`. /// Used to compute `affinity_hit` before calling `try_claim`. pub fn has_session_for(&self, scope: &SessionScope) -> bool { self.agents.iter().any(|slot| { slot.as_ref() - .map(|a| a.state.sessions.contains_key(scope)) + .map(|a| self.agent_owns_scope(a, scope)) .unwrap_or(false) }) } + fn agent_owns_scope(&self, agent: &OwnedAgent, scope: &SessionScope) -> bool { + let Some(owner) = self.session_owners.get(scope) else { + return false; + }; + owner.agent_index == agent.index + && agent.state.scope_owner_generations.get(scope) == Some(&owner.generation) + && agent.state.sessions.contains_key(scope) + } + + /// Earliest scheduled wake for a currently held scope that can claim a + /// worker. A worker return wakes the main loop independently, so arming an + /// already-expired timer while every slot is checked out would only spin. + pub(crate) fn next_hold_deadline(&self, timeout: Duration) -> Option { + if !self.any_idle() { + return None; + } + self.held_since + .values() + .map(|held_since| *held_since + timeout) + .min() + } + + /// Sleep until a held scope's scheduled wake, or remain pending when no + /// scope is held. This is the future polled directly by the main + /// `select!`, kept here so paused-time tests exercise the production seam. + pub(crate) async fn wait_for_hold_deadline(deadline: Option) { + match deadline { + Some(deadline) => tokio::time::sleep_until(deadline).await, + None => std::future::pending().await, + } + } + /// Count of agents that are alive: idle OR checked out (have a task_map entry). /// /// Used to detect when all agents have exited so the caller can respawn. @@ -7461,9 +7578,16 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" agent.state.sessions.insert(ta.clone(), "sess-a".into()); agent.state.sessions.insert(tb.clone(), "sess-b".into()); let mut pool = AgentPool::from_slots(vec![Some(agent)]); - pool.record_scope_owner(ta.clone(), 0); - pool.record_scope_owner(tb.clone(), 0); - let now = std::time::Instant::now(); + let ta_generation = pool.record_scope_owner(ta.clone(), 0); + let tb_generation = pool.record_scope_owner(tb.clone(), 0); + let agent = pool.agents[0].as_mut().expect("idle test agent"); + agent + .state + .set_scope_owner_generation(ta.clone(), ta_generation); + agent + .state + .set_scope_owner_generation(tb.clone(), tb_generation); + let now = tokio::time::Instant::now(); pool.held_since.insert(ta.clone(), now); pool.held_since.insert(tb.clone(), now); @@ -7515,12 +7639,12 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" /// An idle agent (slot 0) holding a provider session for `scope`, so /// `has_session_for(scope)` is true. - async fn idle_agent_with_session(scope: SessionScope) -> OwnedAgent { + async fn idle_agent_with_session(index: usize, scope: SessionScope) -> OwnedAgent { let acp = AcpClient::spawn("bash", &["-c".into(), "sleep 10".into()], &[], false) .await .expect("spawn dummy ACP"); let mut agent = OwnedAgent { - index: 0, + index, acp, state: SessionState::default(), model_capabilities: None, @@ -7608,7 +7732,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" }, ]; - let base = std::time::Instant::now(); + let base = tokio::time::Instant::now(); for row in rows { let ch = Uuid::new_v4(); let scope = if row.is_thread { @@ -7617,12 +7741,19 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" conv(ch) }; let slots = if row.has_session { - vec![Some(idle_agent_with_session(scope.clone()).await)] + vec![Some(idle_agent_with_session(0, scope.clone()).await)] } else { vec![] }; let mut pool = AgentPool::from_slots(slots); - if row.owner_busy { + if row.has_session { + let generation = pool.record_scope_owner(scope.clone(), 0); + pool.agents[0] + .as_mut() + .expect("idle test agent") + .state + .set_scope_owner_generation(scope.clone(), generation); + } else if row.owner_busy { pool.record_scope_owner(scope.clone(), 1); mark_agent_busy(&mut pool, 1, thread_scope(ch, &"b".repeat(64))); } @@ -7648,8 +7779,12 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" _ => panic!("{}: expected {:?}, got {decision:?}", row.name, row.expect), } - // held_since holds the scope only while a Hold is outstanding. - if matches!(decision, HoldDecision::Hold { .. }) { + // An expired hold remains sticky until a worker is successfully + // claimed; only immediate dispatch clears it here. + if matches!( + decision, + HoldDecision::Hold { .. } | HoldDecision::ForkAfterHold { .. } + ) { assert!( pool.held_since.contains_key(&scope), "{}: hold stamps held_since", @@ -7665,6 +7800,159 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" } } + #[tokio::test(start_paused = true)] + async fn held_scope_deadline_wakes_a_quiet_dispatch_loop() { + let channel_id = Uuid::new_v4(); + let scope = thread_scope(channel_id, &"a".repeat(64)); + let idle_scope = thread_scope(channel_id, &"c".repeat(64)); + let idle_agent = idle_agent_with_session(0, idle_scope).await; + let mut pool = AgentPool::from_slots(vec![Some(idle_agent)]); + pool.record_scope_owner(scope.clone(), 1); + mark_agent_busy(&mut pool, 1, thread_scope(channel_id, &"b".repeat(64))); + + let started = tokio::time::Instant::now(); + assert!(matches!( + pool.hold_decision(&scope, started, HOLD_BUSY_OWNER_TIMEOUT), + HoldDecision::Hold { .. } + )); + let deadline = pool + .next_hold_deadline(HOLD_BUSY_OWNER_TIMEOUT) + .expect("held scope schedules an independent wake"); + let wake = AgentPool::wait_for_hold_deadline(Some(deadline)); + tokio::pin!(wake); + + tokio::time::advance(HOLD_BUSY_OWNER_TIMEOUT - Duration::from_millis(1)).await; + assert!( + tokio::time::timeout(Duration::ZERO, &mut wake) + .await + .is_err(), + "quiet loop stays asleep before deadline" + ); + tokio::time::advance(Duration::from_millis(1)).await; + wake.await; + + assert!(matches!( + pool.hold_decision(&scope, tokio::time::Instant::now(), HOLD_BUSY_OWNER_TIMEOUT), + HoldDecision::ForkAfterHold { .. } + )); + } + + #[tokio::test] + async fn expired_hold_survives_pool_exhaustion_until_a_worker_is_claimable() { + let channel_id = Uuid::new_v4(); + let scope = thread_scope(channel_id, &"a".repeat(64)); + let mut pool = AgentPool::from_slots(vec![None]); + pool.record_scope_owner(scope.clone(), 1); + mark_agent_busy(&mut pool, 1, thread_scope(channel_id, &"b".repeat(64))); + + let started = tokio::time::Instant::now(); + assert!(matches!( + pool.hold_decision(&scope, started, HOLD_BUSY_OWNER_TIMEOUT), + HoldDecision::Hold { .. } + )); + assert!(matches!( + pool.hold_decision( + &scope, + started + HOLD_BUSY_OWNER_TIMEOUT, + HOLD_BUSY_OWNER_TIMEOUT + ), + HoldDecision::ForkAfterHold { .. } + )); + assert!( + pool.held_since.contains_key(&scope), + "failed claim must not restart the timeout" + ); + assert_eq!( + pool.next_hold_deadline(HOLD_BUSY_OWNER_TIMEOUT), + None, + "an expired hold cannot spin while all workers are checked out" + ); + + let idle_scope = thread_scope(channel_id, &"c".repeat(64)); + pool.agents[0] = Some(idle_agent_with_session(0, idle_scope).await); + assert_eq!( + pool.next_hold_deadline(HOLD_BUSY_OWNER_TIMEOUT), + Some(started + HOLD_BUSY_OWNER_TIMEOUT), + "worker availability immediately re-arms the expired deadline" + ); + assert!(matches!( + pool.hold_decision( + &scope, + started + HOLD_BUSY_OWNER_TIMEOUT + Duration::from_secs(1), + HOLD_BUSY_OWNER_TIMEOUT + ), + HoldDecision::ForkAfterHold { .. } + )); + pool.clear_hold(&scope); + assert!(!pool.held_since.contains_key(&scope)); + } + + #[tokio::test] + async fn forked_scope_discards_stale_session_when_busy_owner_returns() { + let channel_id = Uuid::new_v4(); + let scope = thread_scope(channel_id, &"a".repeat(64)); + let busy_scope = thread_scope(channel_id, &"b".repeat(64)); + let old_owner = idle_agent_with_session(0, scope.clone()).await; + let replacement = idle_agent_with_session(1, busy_scope.clone()).await; + let mut pool = AgentPool::from_slots(vec![Some(old_owner), Some(replacement)]); + + let mut old_owner = pool.try_claim(None).expect("claim worker 0"); + let old_generation = pool.record_scope_owner(scope.clone(), old_owner.index); + old_owner + .state + .set_scope_owner_generation(scope.clone(), old_generation); + mark_agent_busy(&mut pool, old_owner.index, busy_scope); + + let started = tokio::time::Instant::now(); + assert!(matches!( + pool.hold_decision(&scope, started, HOLD_BUSY_OWNER_TIMEOUT), + HoldDecision::Hold { .. } + )); + assert!(matches!( + pool.hold_decision( + &scope, + started + HOLD_BUSY_OWNER_TIMEOUT, + HOLD_BUSY_OWNER_TIMEOUT + ), + HoldDecision::ForkAfterHold { .. } + )); + + let mut replacement = pool + .try_claim(Some(&scope)) + .expect("idle worker receives forked scope"); + pool.clear_hold(&scope); + assert_eq!(replacement.index, 1); + replacement + .state + .sessions + .insert(scope.clone(), "fresh-session".into()); + let fresh_generation = pool.record_scope_owner(scope.clone(), replacement.index); + replacement + .state + .set_scope_owner_generation(scope.clone(), fresh_generation); + + // Both turns return. Slot order must not make worker 0's old provider + // context claimable after worker 1 became the authoritative owner. + pool.return_agent(replacement); + pool.task_map + .retain(|_, meta| meta.agent_index != old_owner.index); + pool.return_agent(old_owner); + assert!( + !pool.agents[0] + .as_ref() + .expect("worker 0 returned") + .state + .sessions + .contains_key(&scope), + "return cleanup removes the old provider session" + ); + + let claimed = pool + .try_claim(Some(&scope)) + .expect("authoritative owner remains claimable"); + assert_eq!(claimed.index, 1, "next turn resumes the forked session"); + } + #[test] fn test_rotate_after_natural_completion_invalidates_channel_state() { let (mut s, ch_a, ch_b) = make_state(); @@ -10440,7 +10728,7 @@ done"# pool.invalidate_scope_session(&scopes[1]); pool.record_scope_owner(scopes[0].clone(), 0); pool.held_since - .insert(scopes[0].clone(), std::time::Instant::now()); + .insert(scopes[0].clone(), tokio::time::Instant::now()); assert_eq!( pool.switch_idle_agent_model(channel_id, "model-b", Some("pick".into())), IdleSwitchResult::Switched, diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index b2203f62f9d..a419b78178d 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -782,6 +782,22 @@ impl EventQueue { self.queues.len() } + /// Whether `scope` still has work that can be reconstructed into a batch. + /// + /// Busy-owner hold timestamps are derived from pending queue state. Queue + /// cap eviction can retire a scope without going through a pool cleanup + /// path, so the dispatch loop uses this seam to prune orphaned holds before + /// scheduling their deadline wakeups. + pub(crate) fn has_pending_scope(&self, scope: &SessionScope) -> bool { + self.queues + .get(scope) + .is_some_and(|queue| !queue.is_empty()) + || self + .cancelled_batches + .get(scope) + .is_some_and(|events| !events.is_empty()) + } + /// Number of queued events for a specific scope (or channel, treated as its /// conversation scope). Test-only. #[cfg(test)]