mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
feat(agents): add active turn indicators to Agents Menu (#1005)
Signed-off-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <wpfleger@squareup.com> Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub1fgdl5qqnh3k3f2xkqrvt7cujalhm623x4s7fdjdj5yrtp5fzjl9qrjpucw <4a1bfa0013bc6d14a8d600d8bf6392efefbd2a26ac3c96c9b2a106b0d12297ca@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
co-authored by
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
npub1fgdl5qqnh3k3f2xkqrvt7cujalhm623x4s7fdjdj5yrtp5fzjl9qrjpucw
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
npub1fgdl5qqnh3k3f2xkqrvt7cujalhm623x4s7fdjdj5yrtp5fzjl9qrjpucw
parent
39d9aa8260
commit
7983bf6751
@@ -253,6 +253,16 @@ impl AcpClient {
|
||||
self.observer_context = context;
|
||||
}
|
||||
|
||||
/// Return a clone of the observer handle, if attached.
|
||||
pub(crate) fn observer_handle(&self) -> Option<ObserverHandle> {
|
||||
self.observer.clone()
|
||||
}
|
||||
|
||||
/// Return the pool slot index for this agent process.
|
||||
pub(crate) fn observer_agent_index(&self) -> Option<usize> {
|
||||
self.observer_agent_index
|
||||
}
|
||||
|
||||
/// Emit a semantic event to the local observer feed, if enabled.
|
||||
pub fn observe(&self, kind: impl Into<String>, payload: serde_json::Value) {
|
||||
if let Some(observer) = &self.observer {
|
||||
|
||||
@@ -677,6 +677,16 @@ pub async fn run_prompt_task(
|
||||
.unwrap_or_default();
|
||||
let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone());
|
||||
|
||||
// ── Turn completion guard ─────────────────────────────────────────────
|
||||
// Emits `turn_completed` on any exit path. Captures observer handle and
|
||||
// metadata now, before the agent is moved into PromptResult.
|
||||
let _turn_guard = TurnCompletionGuard::new(
|
||||
agent.acp.observer_handle(),
|
||||
agent.acp.observer_agent_index(),
|
||||
observer_channel_id,
|
||||
turn_id.clone(),
|
||||
);
|
||||
|
||||
let (session_id, is_new_session) = match &source {
|
||||
PromptSource::Channel(cid) => {
|
||||
if let Some(sid) = agent.state.sessions.get(cid) {
|
||||
@@ -1965,6 +1975,49 @@ impl Drop for ReactionGuard {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Turn completion scope guard ──────────────────────────────────────────────
|
||||
// Emits a `turn_completed` observer event on drop, covering ALL exit paths
|
||||
// (success, error, timeout, cancel, panic) from `run_prompt_task`. Captures
|
||||
// observer handle and metadata at creation time so it remains valid even after
|
||||
// the agent is moved into `PromptResult`.
|
||||
|
||||
struct TurnCompletionGuard {
|
||||
observer: Option<observer::ObserverHandle>,
|
||||
agent_index: Option<usize>,
|
||||
channel_id: Option<uuid::Uuid>,
|
||||
turn_id: String,
|
||||
}
|
||||
|
||||
impl TurnCompletionGuard {
|
||||
fn new(
|
||||
observer: Option<observer::ObserverHandle>,
|
||||
agent_index: Option<usize>,
|
||||
channel_id: Option<uuid::Uuid>,
|
||||
turn_id: String,
|
||||
) -> Self {
|
||||
Self {
|
||||
observer,
|
||||
agent_index,
|
||||
channel_id,
|
||||
turn_id,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TurnCompletionGuard {
|
||||
fn drop(&mut self) {
|
||||
if let Some(observer) = self.observer.take() {
|
||||
let context = observer::context_for(self.channel_id, None, Some(self.turn_id.clone()));
|
||||
observer.emit(
|
||||
"turn_completed",
|
||||
self.agent_index,
|
||||
&context,
|
||||
serde_json::json!({}),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const REACTION_SEEN: &str = "👀";
|
||||
const REACTION_WORKING: &str = "💬";
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ export default defineConfig({
|
||||
"**/channel-star-screenshots.spec.ts",
|
||||
"**/channel-controls-screenshots.spec.ts",
|
||||
"**/team-management-screenshots.spec.ts",
|
||||
"**/active-turn-screenshots.spec.ts",
|
||||
"**/file-attachment.spec.ts",
|
||||
"**/video-attachment.spec.ts",
|
||||
"**/mentions.spec.ts",
|
||||
|
||||
@@ -0,0 +1,447 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { describe, it, beforeEach } from "node:test";
|
||||
|
||||
import {
|
||||
syncAgentTurnsFromEvents,
|
||||
getActiveChannelsForAgent,
|
||||
resetActiveAgentTurnsStore,
|
||||
subscribeActiveAgentTurns,
|
||||
} from "./activeAgentTurnsStore.ts";
|
||||
|
||||
const AGENT =
|
||||
"abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234";
|
||||
|
||||
function makeEvent(overrides) {
|
||||
return {
|
||||
seq: 1,
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
kind: "turn_started",
|
||||
agentIndex: 0,
|
||||
channelId: "chan-1",
|
||||
sessionId: "sess-1",
|
||||
turnId: "turn-1",
|
||||
payload: null,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe("activeAgentTurnsStore", () => {
|
||||
beforeEach(() => {
|
||||
resetActiveAgentTurnsStore();
|
||||
});
|
||||
|
||||
describe("seq filtering", () => {
|
||||
it("processes events with increasing seq", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
]);
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(channels.size, 1);
|
||||
assert.ok(channels.has("c1"));
|
||||
});
|
||||
|
||||
it("skips events at or below the watermark", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 5, turnId: "t1", channelId: "c1" }),
|
||||
]);
|
||||
// Try to process an older event — should be ignored
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 3, turnId: "t2", channelId: "c2" }),
|
||||
]);
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(channels.size, 1);
|
||||
assert.ok(channels.has("c1"));
|
||||
assert.ok(!channels.has("c2"));
|
||||
});
|
||||
|
||||
it("skips duplicate seq", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
]);
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t2", channelId: "c2" }),
|
||||
]);
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(channels.size, 1);
|
||||
assert.ok(channels.has("c1"));
|
||||
});
|
||||
});
|
||||
|
||||
describe("seq restart detection", () => {
|
||||
it("processes post-restart events whose timestamp climbs past the watermark", () => {
|
||||
// Process events up to seq 50.
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 50,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
]);
|
||||
assert.equal(getActiveChannelsForAgent(AGENT).size, 1);
|
||||
|
||||
// Agent restarts — seq resets to 1, but wall-clock timestamp keeps
|
||||
// climbing. The composite watermark accepts it on timestamp alone.
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t2",
|
||||
channelId: "c2",
|
||||
timestamp: "2024-01-01T00:01:00Z",
|
||||
}),
|
||||
]);
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
assert.ok(channels.has("c2"), "post-restart event should be processed");
|
||||
});
|
||||
|
||||
it("processes subsequent events after restart", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 100,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
]);
|
||||
|
||||
// Restart: seq goes 1, 2, 3 with climbing timestamps.
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t2",
|
||||
channelId: "c2",
|
||||
timestamp: "2024-01-01T00:01:00Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
turnId: "t3",
|
||||
channelId: "c3",
|
||||
timestamp: "2024-01-01T00:01:01Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 3,
|
||||
kind: "turn_completed",
|
||||
turnId: "t2",
|
||||
channelId: "c2",
|
||||
timestamp: "2024-01-01T00:01:02Z",
|
||||
}),
|
||||
]);
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
// t1 still active (not ended), t2 ended, t3 still active.
|
||||
assert.ok(channels.has("c1"));
|
||||
assert.ok(!channels.has("c2"));
|
||||
assert.ok(channels.has("c3"));
|
||||
});
|
||||
});
|
||||
|
||||
describe("eviction at MAX_TURNS_PER_AGENT", () => {
|
||||
it("evicts oldest turn when exceeding 4 concurrent turns", () => {
|
||||
const events = [];
|
||||
for (let i = 1; i <= 5; i++) {
|
||||
events.push(
|
||||
makeEvent({
|
||||
seq: i,
|
||||
turnId: `t${i}`,
|
||||
channelId: `c${i}`,
|
||||
timestamp: `2024-01-01T00:0${i}:00Z`,
|
||||
}),
|
||||
);
|
||||
}
|
||||
syncAgentTurnsFromEvents(AGENT, events);
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
// Should have evicted c1 (oldest) to make room for c5
|
||||
assert.equal(channels.size, 4);
|
||||
assert.ok(!channels.has("c1"), "oldest turn should be evicted");
|
||||
assert.ok(channels.has("c2"));
|
||||
assert.ok(channels.has("c5"));
|
||||
});
|
||||
});
|
||||
|
||||
describe("endTurn turnId-vs-channelId fallback", () => {
|
||||
it("ends turn by turnId when provided", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_completed",
|
||||
turnId: "t1",
|
||||
channelId: null,
|
||||
}),
|
||||
]);
|
||||
assert.equal(getActiveChannelsForAgent(AGENT).size, 0);
|
||||
});
|
||||
|
||||
it("falls back to channelId when turnId is null", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_completed",
|
||||
turnId: null,
|
||||
channelId: "c1",
|
||||
}),
|
||||
]);
|
||||
assert.equal(getActiveChannelsForAgent(AGENT).size, 0);
|
||||
});
|
||||
|
||||
it("does nothing when both turnId and channelId are null", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_completed",
|
||||
turnId: null,
|
||||
channelId: null,
|
||||
}),
|
||||
]);
|
||||
// Turn should still be active — no way to identify which to end
|
||||
assert.equal(getActiveChannelsForAgent(AGENT).size, 1);
|
||||
});
|
||||
|
||||
it("channelId fallback removes only one matching turn", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
makeEvent({ seq: 2, turnId: "t2", channelId: "c1" }),
|
||||
makeEvent({
|
||||
seq: 3,
|
||||
kind: "turn_completed",
|
||||
turnId: null,
|
||||
channelId: "c1",
|
||||
}),
|
||||
]);
|
||||
// Only one of the two turns in c1 should be removed
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(channels.size, 1);
|
||||
assert.ok(channels.has("c1"));
|
||||
});
|
||||
});
|
||||
|
||||
describe("listener notifications", () => {
|
||||
it("notifies on turn_started", () => {
|
||||
let called = 0;
|
||||
const unsub = subscribeActiveAgentTurns(() => {
|
||||
called++;
|
||||
});
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
]);
|
||||
assert.ok(called > 0);
|
||||
unsub();
|
||||
});
|
||||
|
||||
it("notifies on turn_completed", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
]);
|
||||
let called = 0;
|
||||
const unsub = subscribeActiveAgentTurns(() => {
|
||||
called++;
|
||||
});
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 2, kind: "turn_completed", turnId: "t1" }),
|
||||
]);
|
||||
assert.ok(called > 0);
|
||||
unsub();
|
||||
});
|
||||
});
|
||||
|
||||
describe("replay idempotency", () => {
|
||||
it("replaying the same buffer produces no additional state change or notifications", () => {
|
||||
const buffer = [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
turnId: "t2",
|
||||
channelId: "c2",
|
||||
timestamp: "2024-01-01T00:00:01Z",
|
||||
}),
|
||||
];
|
||||
|
||||
// Initial pass.
|
||||
syncAgentTurnsFromEvents(AGENT, buffer);
|
||||
const afterFirst = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(afterFirst.size, 2);
|
||||
|
||||
// Subscribe, then replay the identical buffer.
|
||||
let notified = 0;
|
||||
const unsub = subscribeActiveAgentTurns(() => {
|
||||
notified++;
|
||||
});
|
||||
syncAgentTurnsFromEvents(AGENT, buffer);
|
||||
unsub();
|
||||
|
||||
assert.equal(notified, 0, "replay must not notify listeners");
|
||||
const afterReplay = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(
|
||||
afterReplay,
|
||||
afterFirst,
|
||||
"replay must not change turn state (stable reference)",
|
||||
);
|
||||
});
|
||||
|
||||
it("post-restart replay does not reprocess seen events or resurrect evicted turns", () => {
|
||||
// Start a turn, then complete it (turn evicted).
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_completed",
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:01Z",
|
||||
}),
|
||||
]);
|
||||
assert.equal(getActiveChannelsForAgent(AGENT).size, 0);
|
||||
|
||||
// Agent restarts. The harness replays its buffer with seq reset to 1,
|
||||
// but the original event timestamps (older than the watermark) are
|
||||
// unchanged. The start event must NOT resurrect the evicted turn.
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_completed",
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:01Z",
|
||||
}),
|
||||
]);
|
||||
assert.equal(
|
||||
getActiveChannelsForAgent(AGENT).size,
|
||||
0,
|
||||
"stale replayed start must not resurrect an evicted turn",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("replayed eviction safety", () => {
|
||||
it("replayed stale turn_error with null turnId does not kill the live turn", () => {
|
||||
// A turn errors out (harness emits turn_error with a null turnId), then a
|
||||
// fresh turn starts in the same channel.
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_error",
|
||||
turnId: null,
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:01Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 3,
|
||||
turnId: "t2",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:02Z",
|
||||
}),
|
||||
]);
|
||||
assert.equal(getActiveChannelsForAgent(AGENT).size, 1);
|
||||
|
||||
// The full buffer is replayed on the next observer event. The stale
|
||||
// turn_error (below the watermark) must NOT re-run its channel-match
|
||||
// fallback and delete the live turn t2.
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_error",
|
||||
turnId: null,
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:01Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 3,
|
||||
turnId: "t2",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:02Z",
|
||||
}),
|
||||
]);
|
||||
const channels = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(
|
||||
channels.size,
|
||||
1,
|
||||
"replayed stale turn_error must not delete the live turn",
|
||||
);
|
||||
assert.ok(channels.has("c1"));
|
||||
});
|
||||
|
||||
it("replaying evictions fires no spurious listener notifications", () => {
|
||||
const buffer = [
|
||||
makeEvent({
|
||||
seq: 1,
|
||||
turnId: "t1",
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:00Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 2,
|
||||
kind: "turn_error",
|
||||
turnId: null,
|
||||
channelId: "c1",
|
||||
timestamp: "2024-01-01T00:00:01Z",
|
||||
}),
|
||||
makeEvent({
|
||||
seq: 3,
|
||||
kind: "agent_panic",
|
||||
turnId: null,
|
||||
channelId: "c2",
|
||||
timestamp: "2024-01-01T00:00:02Z",
|
||||
}),
|
||||
];
|
||||
|
||||
// Initial pass processes the buffer.
|
||||
syncAgentTurnsFromEvents(AGENT, buffer);
|
||||
|
||||
// Subscribe, then replay the identical buffer. Every event is below the
|
||||
// watermark, so the replay must be a complete no-op.
|
||||
let notified = 0;
|
||||
const unsub = subscribeActiveAgentTurns(() => {
|
||||
notified++;
|
||||
});
|
||||
syncAgentTurnsFromEvents(AGENT, buffer);
|
||||
unsub();
|
||||
|
||||
assert.equal(notified, 0, "replayed evictions must not notify listeners");
|
||||
});
|
||||
});
|
||||
|
||||
describe("getActiveChannelsForAgent", () => {
|
||||
it("returns EMPTY_SET for null/undefined pubkey", () => {
|
||||
assert.equal(getActiveChannelsForAgent(null).size, 0);
|
||||
assert.equal(getActiveChannelsForAgent(undefined).size, 0);
|
||||
});
|
||||
|
||||
it("returns stable reference when unchanged", () => {
|
||||
syncAgentTurnsFromEvents(AGENT, [
|
||||
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
||||
]);
|
||||
const ref1 = getActiveChannelsForAgent(AGENT);
|
||||
const ref2 = getActiveChannelsForAgent(AGENT);
|
||||
assert.equal(ref1, ref2, "should return cached reference");
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,291 @@
|
||||
import * as React from "react";
|
||||
|
||||
import {
|
||||
subscribeAgentObserverStore,
|
||||
getAgentObserverSnapshot,
|
||||
compareObserverEvents,
|
||||
} from "@/features/agents/observerRelayStore";
|
||||
import { normalizePubkey } from "@/shared/lib/pubkey";
|
||||
import type { ObserverEvent } from "./ui/agentSessionTypes";
|
||||
|
||||
/** Remove a turn entirely after 90s of no activity. */
|
||||
const REMOVE_AFTER_MS = 90_000;
|
||||
/** Maximum concurrent active turns tracked per agent (matches pool size). */
|
||||
const MAX_TURNS_PER_AGENT = 4;
|
||||
/** Interval for pruning stale/expired turns. */
|
||||
const PRUNE_INTERVAL_MS = 5_000;
|
||||
|
||||
type ActiveTurn = {
|
||||
turnId: string;
|
||||
channelId: string;
|
||||
startedAt: number;
|
||||
lastActivityAt: number;
|
||||
};
|
||||
|
||||
// Module-level state: agentPubkey → turnId → ActiveTurn
|
||||
const activeTurnsByAgent = new Map<string, Map<string, ActiveTurn>>();
|
||||
const listeners = new Set<() => void>();
|
||||
|
||||
// Cached snapshots for useSyncExternalStore reference stability.
|
||||
// Only regenerated when the underlying turn map for an agent actually changes.
|
||||
const cachedChannelSets = new Map<string, Set<string>>();
|
||||
|
||||
// Composite watermark per agent: the newest observer event processed, by
|
||||
// (timestamp, seq) ordering. An event is processed only if it is strictly
|
||||
// newer than this — making full-buffer replays idempotent and post-restart
|
||||
// streams (seq resets to 1, timestamp keeps climbing) handled for free.
|
||||
const lastProcessed = new Map<string, ObserverEvent>();
|
||||
|
||||
let pruneInterval: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
function invalidateCache(agentKey: string) {
|
||||
cachedChannelSets.delete(agentKey);
|
||||
}
|
||||
|
||||
function notifyListeners() {
|
||||
for (const listener of listeners) {
|
||||
listener();
|
||||
}
|
||||
}
|
||||
|
||||
function startTurn(
|
||||
agentPubkey: string,
|
||||
channelId: string,
|
||||
turnId: string,
|
||||
timestamp: string,
|
||||
) {
|
||||
const key = normalizePubkey(agentPubkey);
|
||||
let agentTurns = activeTurnsByAgent.get(key);
|
||||
if (!agentTurns) {
|
||||
agentTurns = new Map();
|
||||
activeTurnsByAgent.set(key, agentTurns);
|
||||
}
|
||||
|
||||
// Cap at MAX_TURNS_PER_AGENT — evict oldest if exceeded
|
||||
if (agentTurns.size >= MAX_TURNS_PER_AGENT && !agentTurns.has(turnId)) {
|
||||
let oldestKey: string | null = null;
|
||||
let oldestTime = Number.POSITIVE_INFINITY;
|
||||
for (const [tid, turn] of agentTurns) {
|
||||
if (turn.startedAt < oldestTime) {
|
||||
oldestTime = turn.startedAt;
|
||||
oldestKey = tid;
|
||||
}
|
||||
}
|
||||
if (oldestKey) {
|
||||
agentTurns.delete(oldestKey);
|
||||
}
|
||||
}
|
||||
|
||||
const now = Date.parse(timestamp) || Date.now();
|
||||
agentTurns.set(turnId, {
|
||||
turnId,
|
||||
channelId,
|
||||
startedAt: now,
|
||||
lastActivityAt: now,
|
||||
});
|
||||
invalidateCache(key);
|
||||
}
|
||||
|
||||
function recordActivity(agentPubkey: string, turnId: string | null) {
|
||||
if (!turnId) return;
|
||||
const key = normalizePubkey(agentPubkey);
|
||||
const agentTurns = activeTurnsByAgent.get(key);
|
||||
if (!agentTurns) return;
|
||||
const turn = agentTurns.get(turnId);
|
||||
if (turn) {
|
||||
turn.lastActivityAt = Date.now();
|
||||
}
|
||||
}
|
||||
|
||||
function endTurn(
|
||||
agentPubkey: string,
|
||||
turnId: string | null,
|
||||
channelId: string | null,
|
||||
) {
|
||||
const key = normalizePubkey(agentPubkey);
|
||||
const agentTurns = activeTurnsByAgent.get(key);
|
||||
if (!agentTurns) return;
|
||||
|
||||
if (turnId) {
|
||||
agentTurns.delete(turnId);
|
||||
} else if (channelId) {
|
||||
// Fallback: remove by channelId if turnId not available
|
||||
for (const [tid, turn] of agentTurns) {
|
||||
if (turn.channelId === channelId) {
|
||||
agentTurns.delete(tid);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (agentTurns.size === 0) {
|
||||
activeTurnsByAgent.delete(key);
|
||||
}
|
||||
invalidateCache(key);
|
||||
}
|
||||
|
||||
function pruneExpired() {
|
||||
const now = Date.now();
|
||||
let changed = false;
|
||||
for (const [agentKey, agentTurns] of activeTurnsByAgent) {
|
||||
for (const [turnId, turn] of agentTurns) {
|
||||
if (now - turn.lastActivityAt > REMOVE_AFTER_MS) {
|
||||
agentTurns.delete(turnId);
|
||||
invalidateCache(agentKey);
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
if (agentTurns.size === 0) {
|
||||
activeTurnsByAgent.delete(agentKey);
|
||||
}
|
||||
}
|
||||
if (changed) {
|
||||
notifyListeners();
|
||||
}
|
||||
}
|
||||
|
||||
// INVARIANT: events must be sorted by (timestamp, seq) ascending.
|
||||
// syncAgentTurnsFromEvents receives sorted arrays from observerRelayStore.
|
||||
// Calling with unsorted events will cause silent data loss.
|
||||
function processEvent(agentPubkey: string, event: ObserverEvent) {
|
||||
const key = normalizePubkey(agentPubkey);
|
||||
|
||||
// Gate every event kind on the watermark uniformly: process only events
|
||||
// strictly newer than the last one seen for this agent. With sorted buffers
|
||||
// (the documented invariant), this makes full-buffer replays a complete
|
||||
// no-op. Evictions must be gated too — replaying a stale turn_error/
|
||||
// agent_panic (emitted with a null turnId) would otherwise fall back to
|
||||
// deleting the first turn in the channel, killing the live turn. Resurrection
|
||||
// is not a concern: it would require reprocessing a stale start, which the
|
||||
// watermark already blocks.
|
||||
const last = lastProcessed.get(key);
|
||||
if (last && compareObserverEvents(event, last) <= 0) {
|
||||
return;
|
||||
}
|
||||
lastProcessed.set(key, event);
|
||||
|
||||
switch (event.kind) {
|
||||
case "turn_started":
|
||||
if (event.channelId) {
|
||||
startTurn(
|
||||
agentPubkey,
|
||||
event.channelId,
|
||||
event.turnId ?? `seq-${event.seq}`,
|
||||
event.timestamp,
|
||||
);
|
||||
notifyListeners();
|
||||
}
|
||||
break;
|
||||
case "turn_completed":
|
||||
case "turn_error":
|
||||
case "agent_panic":
|
||||
endTurn(agentPubkey, event.turnId ?? null, event.channelId ?? null);
|
||||
notifyListeners();
|
||||
break;
|
||||
case "acp_read":
|
||||
case "acp_write":
|
||||
recordActivity(agentPubkey, event.turnId ?? null);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
function ensurePruneInterval() {
|
||||
if (pruneInterval) return;
|
||||
pruneInterval = setInterval(pruneExpired, PRUNE_INTERVAL_MS);
|
||||
}
|
||||
|
||||
function stopPruneInterval() {
|
||||
if (pruneInterval) {
|
||||
clearInterval(pruneInterval);
|
||||
pruneInterval = null;
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Public API ──────────────────────────────────────────────────────────────
|
||||
|
||||
export function subscribeActiveAgentTurns(listener: () => void) {
|
||||
listeners.add(listener);
|
||||
if (listeners.size === 1) {
|
||||
ensurePruneInterval();
|
||||
}
|
||||
return () => {
|
||||
listeners.delete(listener);
|
||||
if (listeners.size === 0) {
|
||||
stopPruneInterval();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/** Returns the set of channel IDs where the given agent has active turns. */
|
||||
export function getActiveChannelsForAgent(
|
||||
agentPubkey: string | null | undefined,
|
||||
): Set<string> {
|
||||
if (!agentPubkey) return EMPTY_SET;
|
||||
const key = normalizePubkey(agentPubkey);
|
||||
const agentTurns = activeTurnsByAgent.get(key);
|
||||
if (!agentTurns || agentTurns.size === 0) return EMPTY_SET;
|
||||
|
||||
const cached = cachedChannelSets.get(key);
|
||||
if (cached) return cached;
|
||||
|
||||
const result = new Set([...agentTurns.values()].map((t) => t.channelId));
|
||||
cachedChannelSets.set(key, result);
|
||||
return result;
|
||||
}
|
||||
|
||||
const EMPTY_SET: Set<string> = new Set();
|
||||
|
||||
/**
|
||||
* Synchronize the active-turns store with the latest observer events for a
|
||||
* given agent.
|
||||
*/
|
||||
export function syncAgentTurnsFromEvents(
|
||||
agentPubkey: string,
|
||||
events: ObserverEvent[],
|
||||
) {
|
||||
for (const event of events) {
|
||||
processEvent(agentPubkey, event);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hook: returns the set of channel IDs where the given agent is currently working.
|
||||
* Re-renders when the set changes.
|
||||
*/
|
||||
export function useActiveAgentTurns(
|
||||
agentPubkey: string | null | undefined,
|
||||
): Set<string> {
|
||||
const getSnapshot = React.useCallback(
|
||||
() => getActiveChannelsForAgent(agentPubkey),
|
||||
[agentPubkey],
|
||||
);
|
||||
|
||||
return React.useSyncExternalStore(subscribeActiveAgentTurns, getSnapshot);
|
||||
}
|
||||
|
||||
/**
|
||||
* Bridge hook: processes observer events into the active-turns store.
|
||||
* Should be called by a parent component that has access to the observer events.
|
||||
*/
|
||||
export function useActiveAgentTurnsBridge(
|
||||
agents: readonly { pubkey: string; status: string }[],
|
||||
) {
|
||||
React.useEffect(() => {
|
||||
function syncAll() {
|
||||
for (const agent of agents) {
|
||||
if (agent.status !== "running" && agent.status !== "deployed") continue;
|
||||
const snapshot = getAgentObserverSnapshot(agent.pubkey, true);
|
||||
syncAgentTurnsFromEvents(agent.pubkey, snapshot.events);
|
||||
}
|
||||
}
|
||||
|
||||
syncAll();
|
||||
return subscribeAgentObserverStore(syncAll);
|
||||
}, [agents]);
|
||||
}
|
||||
|
||||
export function resetActiveAgentTurnsStore() {
|
||||
activeTurnsByAgent.clear();
|
||||
lastProcessed.clear();
|
||||
cachedChannelSets.clear();
|
||||
notifyListeners();
|
||||
}
|
||||
@@ -115,7 +115,10 @@ function appendAgentEvent(agentPubkey: string, event: ObserverEvent) {
|
||||
notifyListeners();
|
||||
}
|
||||
|
||||
function compareObserverEvents(left: ObserverEvent, right: ObserverEvent) {
|
||||
export function compareObserverEvents(
|
||||
left: ObserverEvent,
|
||||
right: ObserverEvent,
|
||||
) {
|
||||
const leftTime = Date.parse(left.timestamp);
|
||||
const rightTime = Date.parse(right.timestamp);
|
||||
if (Number.isFinite(leftTime) && Number.isFinite(rightTime)) {
|
||||
|
||||
@@ -4,7 +4,8 @@ import { ManagedAgentRow } from "./ManagedAgentRow";
|
||||
|
||||
export type AgentGroupRowsProps = {
|
||||
agents: ManagedAgent[];
|
||||
channelsByPubkey: Record<string, string[]>;
|
||||
channelIdToName: Record<string, string>;
|
||||
channelsByPubkey: Record<string, { id: string; name: string }[]>;
|
||||
isActionPending: boolean;
|
||||
logContent: string | null;
|
||||
logError: Error | null;
|
||||
@@ -23,6 +24,7 @@ export type AgentGroupRowsProps = {
|
||||
|
||||
export function AgentGroupRows({
|
||||
agents,
|
||||
channelIdToName,
|
||||
channelsByPubkey,
|
||||
isActionPending,
|
||||
logContent,
|
||||
@@ -44,6 +46,7 @@ export function AgentGroupRows({
|
||||
{agents.map((agent) => (
|
||||
<ManagedAgentRow
|
||||
agent={agent}
|
||||
channelIdToName={channelIdToName}
|
||||
channelNames={channelsByPubkey[normalizePubkey(agent.pubkey)] ?? []}
|
||||
isActionPending={isActionPending}
|
||||
isLogSelected={selectedLogAgentPubkey === agent.pubkey}
|
||||
|
||||
@@ -7,10 +7,12 @@ import type { ManagedAgent, PresenceStatus } from "@/shared/api/types";
|
||||
const PRESENCE_GRACE_MS = 15_000;
|
||||
|
||||
export function AgentStatusBadge({
|
||||
isWorking,
|
||||
presenceLoaded,
|
||||
presenceStatus,
|
||||
status,
|
||||
}: {
|
||||
isWorking?: boolean;
|
||||
presenceLoaded: boolean;
|
||||
presenceStatus: PresenceStatus | undefined;
|
||||
status: ManagedAgent["status"];
|
||||
@@ -29,11 +31,26 @@ export function AgentStatusBadge({
|
||||
status === "running" &&
|
||||
(!presenceStatus || presenceStatus === "offline");
|
||||
|
||||
const variant = isStarting ? "warning" : isActive ? "default" : "secondary";
|
||||
const variant: "default" | "warning" | "secondary" = isWorking
|
||||
? "default"
|
||||
: isStarting
|
||||
? "warning"
|
||||
: isActive
|
||||
? "default"
|
||||
: "secondary";
|
||||
|
||||
const label = isWorking
|
||||
? "Working"
|
||||
: isStarting
|
||||
? "Starting\u2026"
|
||||
: status.replace(/_/g, " ");
|
||||
|
||||
return (
|
||||
<Badge variant={variant}>
|
||||
{isStarting ? "Starting\u2026" : status.replace(/_/g, " ")}
|
||||
<Badge
|
||||
className={isWorking ? "motion-safe:animate-pulse" : undefined}
|
||||
variant={variant}
|
||||
>
|
||||
{label}
|
||||
</Badge>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -49,6 +49,7 @@ export function AgentsView() {
|
||||
actionErrorMessage={agents.actionErrorMessage}
|
||||
actionNoticeMessage={agents.actionNoticeMessage}
|
||||
agents={agents.managedAgents}
|
||||
channelIdToName={agents.channelIdToName}
|
||||
channelsByPubkey={agents.channelsByPubkey}
|
||||
agentsError={
|
||||
agents.managedAgentsQuery.error instanceof Error
|
||||
|
||||
@@ -15,9 +15,11 @@ import {
|
||||
} from "lucide-react";
|
||||
import { toast } from "sonner";
|
||||
|
||||
import { useAppNavigation } from "@/app/navigation/useAppNavigation";
|
||||
import { PresenceDot } from "@/features/presence/ui/PresenceBadge";
|
||||
import { Badge } from "@/shared/ui/badge";
|
||||
import { AgentStatusBadge } from "@/features/agents/ui/AgentStatusBadge";
|
||||
import { useActiveAgentTurns } from "@/features/agents/activeAgentTurnsStore";
|
||||
import type {
|
||||
ManagedAgent,
|
||||
PresenceLookup,
|
||||
@@ -39,6 +41,7 @@ import { truncatePubkey } from "./agentUi";
|
||||
|
||||
export function ManagedAgentRow({
|
||||
agent,
|
||||
channelIdToName,
|
||||
channelNames,
|
||||
isActionPending,
|
||||
isLogSelected,
|
||||
@@ -56,7 +59,8 @@ export function ManagedAgentRow({
|
||||
onToggleStartOnAppLaunch,
|
||||
}: {
|
||||
agent: ManagedAgent;
|
||||
channelNames: string[];
|
||||
channelIdToName: Record<string, string>;
|
||||
channelNames: { id: string; name: string }[];
|
||||
isActionPending: boolean;
|
||||
isLogSelected: boolean;
|
||||
logContent: string | null;
|
||||
@@ -80,6 +84,15 @@ export function ManagedAgentRow({
|
||||
? (personaLabelsById[agent.personaId] ?? null)
|
||||
: null;
|
||||
const presenceStatus = presenceLookup[agent.pubkey.trim().toLowerCase()];
|
||||
const activeChannelIds = useActiveAgentTurns(agent.pubkey);
|
||||
const activeWorkingChannels = React.useMemo(
|
||||
() =>
|
||||
[...activeChannelIds]
|
||||
.map((id) => ({ id, name: channelIdToName[id] ?? id }))
|
||||
.slice(0, 3),
|
||||
[activeChannelIds, channelIdToName],
|
||||
);
|
||||
const isWorking = activeWorkingChannels.length > 0;
|
||||
const processDetail =
|
||||
agent.pid !== null
|
||||
? `PID ${agent.pid}`
|
||||
@@ -116,6 +129,7 @@ export function ManagedAgentRow({
|
||||
>
|
||||
<div className="grid gap-3 lg:grid-cols-[minmax(0,1.8fr)_minmax(120px,0.8fr)_minmax(0,1.1fr)] lg:gap-4">
|
||||
<AgentSummary
|
||||
activeWorkingChannels={activeWorkingChannels}
|
||||
agent={agent}
|
||||
channelNames={channelNames}
|
||||
isExpandable
|
||||
@@ -125,6 +139,7 @@ export function ManagedAgentRow({
|
||||
/>
|
||||
<StatusBlock
|
||||
friendlyError={friendlyError}
|
||||
isWorking={isWorking}
|
||||
presenceLoaded={presenceLoaded}
|
||||
presenceStatus={presenceStatus}
|
||||
processDetail={processDetail}
|
||||
@@ -137,6 +152,7 @@ export function ManagedAgentRow({
|
||||
<div className="min-w-0 flex-1">
|
||||
<div className="grid gap-3 lg:grid-cols-[minmax(0,1.8fr)_minmax(120px,0.8fr)_minmax(0,1.1fr)] lg:gap-4">
|
||||
<AgentSummary
|
||||
activeWorkingChannels={activeWorkingChannels}
|
||||
agent={agent}
|
||||
channelNames={channelNames}
|
||||
isExpandable={false}
|
||||
@@ -146,6 +162,7 @@ export function ManagedAgentRow({
|
||||
/>
|
||||
<StatusBlock
|
||||
friendlyError={friendlyError}
|
||||
isWorking={isWorking}
|
||||
presenceLoaded={presenceLoaded}
|
||||
presenceStatus={presenceStatus}
|
||||
processDetail={processDetail}
|
||||
@@ -191,6 +208,7 @@ export function ManagedAgentRow({
|
||||
}
|
||||
|
||||
function AgentSummary({
|
||||
activeWorkingChannels,
|
||||
agent,
|
||||
channelNames,
|
||||
isExpandable,
|
||||
@@ -198,13 +216,16 @@ function AgentSummary({
|
||||
personaLabel,
|
||||
presenceStatus,
|
||||
}: {
|
||||
activeWorkingChannels: { id: string; name: string }[];
|
||||
agent: ManagedAgent;
|
||||
channelNames: string[];
|
||||
channelNames: { id: string; name: string }[];
|
||||
isExpandable: boolean;
|
||||
isLogSelected: boolean;
|
||||
personaLabel: string | null;
|
||||
presenceStatus: PresenceStatus | undefined;
|
||||
}) {
|
||||
const { goChannel } = useAppNavigation();
|
||||
|
||||
return (
|
||||
<div className="min-w-0">
|
||||
<div className="flex items-start gap-3">
|
||||
@@ -242,13 +263,34 @@ function AgentSummary({
|
||||
</div>
|
||||
{channelNames.length > 0 ? (
|
||||
<div className="mt-1.5 flex flex-wrap items-center gap-1.5">
|
||||
{channelNames.map((name) => (
|
||||
{channelNames.map((channel) => (
|
||||
<Badge
|
||||
className="normal-case tracking-normal"
|
||||
key={name}
|
||||
className="cursor-pointer normal-case tracking-normal hover:opacity-80"
|
||||
key={channel.id}
|
||||
variant="outline"
|
||||
onClick={(e) => {
|
||||
e.stopPropagation();
|
||||
void goChannel(channel.id);
|
||||
}}
|
||||
>
|
||||
# {name}
|
||||
# {channel.name}
|
||||
</Badge>
|
||||
))}
|
||||
</div>
|
||||
) : null}
|
||||
{activeWorkingChannels.length > 0 ? (
|
||||
<div className="mt-1.5 flex flex-wrap items-center gap-1.5">
|
||||
{activeWorkingChannels.map((channel) => (
|
||||
<Badge
|
||||
className="cursor-pointer motion-safe:animate-pulse normal-case tracking-normal hover:opacity-80"
|
||||
key={`working-${channel.id}`}
|
||||
variant="default"
|
||||
onClick={(e) => {
|
||||
e.stopPropagation();
|
||||
void goChannel(channel.id);
|
||||
}}
|
||||
>
|
||||
Working in #{channel.name}
|
||||
</Badge>
|
||||
))}
|
||||
</div>
|
||||
@@ -261,12 +303,14 @@ function AgentSummary({
|
||||
|
||||
function StatusBlock({
|
||||
friendlyError,
|
||||
isWorking,
|
||||
presenceLoaded,
|
||||
presenceStatus,
|
||||
processDetail,
|
||||
status,
|
||||
}: {
|
||||
friendlyError: ReturnType<typeof friendlyAgentLastError>;
|
||||
isWorking: boolean;
|
||||
presenceLoaded: boolean;
|
||||
presenceStatus: PresenceStatus | undefined;
|
||||
processDetail: string;
|
||||
@@ -278,6 +322,7 @@ function StatusBlock({
|
||||
Status
|
||||
</p>
|
||||
<AgentStatusBadge
|
||||
isWorking={isWorking}
|
||||
presenceLoaded={presenceLoaded}
|
||||
presenceStatus={presenceStatus}
|
||||
status={status}
|
||||
|
||||
@@ -37,7 +37,8 @@ type UnifiedAgentsSectionProps = {
|
||||
actionErrorMessage: string | null;
|
||||
actionNoticeMessage: string | null;
|
||||
agents: ManagedAgent[];
|
||||
channelsByPubkey: Record<string, string[]>;
|
||||
channelIdToName: Record<string, string>;
|
||||
channelsByPubkey: Record<string, { id: string; name: string }[]>;
|
||||
agentsError: Error | null;
|
||||
isActionPending: boolean;
|
||||
isAgentsLoading: boolean;
|
||||
@@ -109,6 +110,7 @@ export function UnifiedAgentsSection(props: UnifiedAgentsSectionProps) {
|
||||
actionErrorMessage,
|
||||
actionNoticeMessage,
|
||||
agents,
|
||||
channelIdToName,
|
||||
channelsByPubkey,
|
||||
agentsError,
|
||||
isActionPending,
|
||||
@@ -177,6 +179,7 @@ export function UnifiedAgentsSection(props: UnifiedAgentsSectionProps) {
|
||||
const isLoading = isAgentsLoading || isPersonasLoading;
|
||||
|
||||
const rowProps = {
|
||||
channelIdToName,
|
||||
channelsByPubkey,
|
||||
isActionPending,
|
||||
logContent,
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
import { useChannelsQuery } from "@/features/channels/hooks";
|
||||
import { usePresenceQuery } from "@/features/presence/hooks";
|
||||
import { useManagedAgentObserverBridge } from "@/features/agents/observerRelayStore";
|
||||
import { useActiveAgentTurnsBridge } from "@/features/agents/activeAgentTurnsStore";
|
||||
import type {
|
||||
Channel,
|
||||
CreateManagedAgentResponse,
|
||||
@@ -64,6 +65,7 @@ export function useManagedAgentActions() {
|
||||
[managedAgentsQuery.data],
|
||||
);
|
||||
useManagedAgentObserverBridge(managedAgents);
|
||||
useActiveAgentTurnsBridge(managedAgents);
|
||||
|
||||
const managedPubkeys = React.useMemo(
|
||||
() => new Set(managedAgents.map((agent) => agent.pubkey)),
|
||||
@@ -78,11 +80,18 @@ export function useManagedAgentActions() {
|
||||
const managedPresenceQuery = usePresenceQuery(managedPubkeyList);
|
||||
|
||||
const channelsByPubkey = React.useMemo(() => {
|
||||
const map: Record<string, string[]> = {};
|
||||
const map: Record<string, { id: string; name: string }[]> = {};
|
||||
// Seed from relay agent profiles (kind:10100 events).
|
||||
for (const ra of relayAgentsQuery.data ?? []) {
|
||||
if (ra.channels.length > 0) {
|
||||
map[normalizePubkey(ra.pubkey)] = ra.channels;
|
||||
// Skip entries missing a channel id rather than falling back to the
|
||||
// name as id — a misaligned channels/channelIds pairing would otherwise
|
||||
// produce a pill that silently navigates to a channel name as if it
|
||||
// were an id.
|
||||
map[normalizePubkey(ra.pubkey)] = ra.channels.flatMap((name, i) => {
|
||||
const id = ra.channelIds[i];
|
||||
return id ? [{ id, name }] : [];
|
||||
});
|
||||
}
|
||||
}
|
||||
// Fill in from channel member lists (kind:39002) for any managed agents
|
||||
@@ -95,14 +104,22 @@ export function useManagedAgentActions() {
|
||||
const key = normalizePubkey(pk);
|
||||
if (!normalizedManaged.has(key)) continue;
|
||||
if (!map[key]) map[key] = [];
|
||||
if (!map[key].includes(ch.name)) {
|
||||
map[key].push(ch.name);
|
||||
if (!map[key].some((entry) => entry.id === ch.id)) {
|
||||
map[key].push({ id: ch.id, name: ch.name });
|
||||
}
|
||||
}
|
||||
}
|
||||
return map;
|
||||
}, [relayAgentsQuery.data, channelsQuery.data, managedAgents]);
|
||||
|
||||
const channelIdToName = React.useMemo(() => {
|
||||
const map: Record<string, string> = {};
|
||||
for (const ch of channelsQuery.data ?? []) {
|
||||
map[ch.id] = ch.name;
|
||||
}
|
||||
return map;
|
||||
}, [channelsQuery.data]);
|
||||
|
||||
// Clear log selection if the agent was removed
|
||||
React.useEffect(() => {
|
||||
if (
|
||||
@@ -323,6 +340,7 @@ export function useManagedAgentActions() {
|
||||
// Derived state
|
||||
managedAgents,
|
||||
managedPubkeys,
|
||||
channelIdToName,
|
||||
channelsByPubkey,
|
||||
isPending,
|
||||
// UI state
|
||||
|
||||
@@ -5,6 +5,7 @@ import { finalizeEvent, getPublicKey } from "nostr-tools/pure";
|
||||
import { parse as yamlParse } from "yaml";
|
||||
|
||||
import type { RelayEvent } from "@/shared/api/types";
|
||||
import { syncAgentTurnsFromEvents } from "@/features/agents/activeAgentTurnsStore";
|
||||
import {
|
||||
CUSTOM_EMOJI_SET_D_TAG,
|
||||
KIND_EMOJI_SET,
|
||||
@@ -595,6 +596,11 @@ declare global {
|
||||
models?: Array<{ id: string; name: string | null }>;
|
||||
denyReason?: string;
|
||||
}) => void;
|
||||
__BUZZ_E2E_SEED_ACTIVE_TURNS__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelId: string;
|
||||
turnId: string;
|
||||
}) => void;
|
||||
__BUZZ_E2E_EMIT_MOCK_READ_STATE__?: (input: {
|
||||
clientId: string;
|
||||
contexts: Record<string, number>;
|
||||
@@ -5868,6 +5874,26 @@ export function maybeInstallE2eTauriMocks() {
|
||||
if (mesh.denyReason !== undefined)
|
||||
mockMeshState.denyReason = mesh.denyReason;
|
||||
};
|
||||
let seedTurnSeq = Date.now();
|
||||
window.__BUZZ_E2E_SEED_ACTIVE_TURNS__ = ({
|
||||
agentPubkey,
|
||||
channelId,
|
||||
turnId,
|
||||
}) => {
|
||||
seedTurnSeq += 1;
|
||||
syncAgentTurnsFromEvents(agentPubkey, [
|
||||
{
|
||||
seq: seedTurnSeq,
|
||||
timestamp: new Date().toISOString(),
|
||||
kind: "turn_started",
|
||||
agentIndex: 0,
|
||||
channelId,
|
||||
sessionId: null,
|
||||
turnId,
|
||||
payload: null,
|
||||
},
|
||||
]);
|
||||
};
|
||||
const meshNodeStatus = (
|
||||
state: "off" | "running",
|
||||
mode: "serve" | "client" | null,
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
import { expect, test } from "@playwright/test";
|
||||
|
||||
import { installMockBridge } from "../helpers/bridge";
|
||||
|
||||
const SHOTS = "test-results/active-turns";
|
||||
|
||||
// Mock agent pubkeys (distinct from the relay agents seeded by default)
|
||||
const AGENT_PAUL = "aa".repeat(32);
|
||||
const AGENT_DUNCAN = "bb".repeat(32);
|
||||
const AGENT_THUFIR = "cc".repeat(32);
|
||||
|
||||
// Mock channel IDs from the e2e bridge
|
||||
const CHANNEL_GENERAL = "9a1657ac-f7aa-5db0-b632-d8bbeb6dfb50";
|
||||
const CHANNEL_ENGINEERING = "1c7e1c02-87bb-5e88-b2da-5a7a9432d0c9";
|
||||
|
||||
async function waitForBridge(page: import("@playwright/test").Page) {
|
||||
await page.waitForFunction(
|
||||
() =>
|
||||
typeof (window as Window & { __BUZZ_E2E_SEED_ACTIVE_TURNS__?: unknown })
|
||||
.__BUZZ_E2E_SEED_ACTIVE_TURNS__ === "function",
|
||||
null,
|
||||
{ timeout: 10_000 },
|
||||
);
|
||||
}
|
||||
|
||||
async function openAgentsView(page: import("@playwright/test").Page) {
|
||||
await page.goto("/", { waitUntil: "domcontentloaded" });
|
||||
await waitForBridge(page);
|
||||
await page.getByTestId("open-agents-view").click();
|
||||
await expect(page.getByTestId("unified-agents-groups")).toBeVisible({
|
||||
timeout: 10_000,
|
||||
});
|
||||
}
|
||||
|
||||
test.describe("active turn indicator screenshots", () => {
|
||||
test.use({ viewport: { width: 1280, height: 720 } });
|
||||
|
||||
test("01 — baseline: agents running but idle", async ({ page }) => {
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
pubkey: AGENT_PAUL,
|
||||
name: "Paul",
|
||||
status: "running",
|
||||
channelNames: ["general", "engineering"],
|
||||
},
|
||||
{
|
||||
pubkey: AGENT_DUNCAN,
|
||||
name: "Duncan",
|
||||
status: "running",
|
||||
channelNames: ["general", "design"],
|
||||
},
|
||||
{
|
||||
pubkey: AGENT_THUFIR,
|
||||
name: "Thufir",
|
||||
status: "stopped",
|
||||
channelNames: [],
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
await openAgentsView(page);
|
||||
|
||||
const agentsSection = page.getByTestId("unified-agents-groups");
|
||||
await expect(agentsSection).toContainText("Paul");
|
||||
await expect(agentsSection).toContainText("Duncan");
|
||||
await expect(agentsSection).toContainText("Thufir");
|
||||
|
||||
await agentsSection.screenshot({
|
||||
path: `${SHOTS}/01-baseline-idle.png`,
|
||||
});
|
||||
});
|
||||
|
||||
test("02 — single agent working in one channel", async ({ page }) => {
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
pubkey: AGENT_PAUL,
|
||||
name: "Paul",
|
||||
status: "running",
|
||||
channelNames: ["general", "engineering"],
|
||||
},
|
||||
{
|
||||
pubkey: AGENT_DUNCAN,
|
||||
name: "Duncan",
|
||||
status: "running",
|
||||
channelNames: ["general", "design"],
|
||||
},
|
||||
{
|
||||
pubkey: AGENT_THUFIR,
|
||||
name: "Thufir",
|
||||
status: "stopped",
|
||||
channelNames: [],
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
await openAgentsView(page);
|
||||
await waitForBridge(page);
|
||||
|
||||
// Seed Paul as actively working in #general
|
||||
await page.evaluate(
|
||||
({ pubkey, channelId }) => {
|
||||
const win = window as Window & {
|
||||
__BUZZ_E2E_SEED_ACTIVE_TURNS__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelId: string;
|
||||
turnId: string;
|
||||
}) => void;
|
||||
};
|
||||
win.__BUZZ_E2E_SEED_ACTIVE_TURNS__?.({
|
||||
agentPubkey: pubkey,
|
||||
channelId,
|
||||
turnId: "turn-001",
|
||||
});
|
||||
},
|
||||
{ pubkey: AGENT_PAUL, channelId: CHANNEL_GENERAL },
|
||||
);
|
||||
|
||||
// Wait for the "Working" badge to appear
|
||||
await expect(page.getByTestId(`managed-agent-${AGENT_PAUL}`)).toContainText(
|
||||
"Working",
|
||||
{ timeout: 5_000 },
|
||||
);
|
||||
|
||||
const agentsSection = page.getByTestId("unified-agents-groups");
|
||||
await agentsSection.screenshot({
|
||||
path: `${SHOTS}/02-single-agent-working.png`,
|
||||
});
|
||||
});
|
||||
|
||||
test("03 — mixed states: one working in 2 channels, one idle, one stopped", async ({
|
||||
page,
|
||||
}) => {
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
pubkey: AGENT_PAUL,
|
||||
name: "Paul",
|
||||
status: "running",
|
||||
channelNames: ["general", "engineering"],
|
||||
},
|
||||
{
|
||||
pubkey: AGENT_DUNCAN,
|
||||
name: "Duncan",
|
||||
status: "running",
|
||||
channelNames: ["general", "design"],
|
||||
},
|
||||
{
|
||||
pubkey: AGENT_THUFIR,
|
||||
name: "Thufir",
|
||||
status: "stopped",
|
||||
channelNames: [],
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
await openAgentsView(page);
|
||||
await waitForBridge(page);
|
||||
|
||||
// Seed Paul as working in both #general and #engineering
|
||||
await page.evaluate(
|
||||
({ pubkey, channels }) => {
|
||||
const win = window as Window & {
|
||||
__BUZZ_E2E_SEED_ACTIVE_TURNS__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelId: string;
|
||||
turnId: string;
|
||||
}) => void;
|
||||
};
|
||||
for (const { channelId, turnId } of channels) {
|
||||
win.__BUZZ_E2E_SEED_ACTIVE_TURNS__?.({
|
||||
agentPubkey: pubkey,
|
||||
channelId,
|
||||
turnId,
|
||||
});
|
||||
}
|
||||
},
|
||||
{
|
||||
pubkey: AGENT_PAUL,
|
||||
channels: [
|
||||
{ channelId: CHANNEL_GENERAL, turnId: "turn-002" },
|
||||
{ channelId: CHANNEL_ENGINEERING, turnId: "turn-003" },
|
||||
],
|
||||
},
|
||||
);
|
||||
|
||||
// Wait for the "Working" indicators to appear
|
||||
await expect(page.getByTestId(`managed-agent-${AGENT_PAUL}`)).toContainText(
|
||||
"Working",
|
||||
{ timeout: 5_000 },
|
||||
);
|
||||
|
||||
const agentsSection = page.getByTestId("unified-agents-groups");
|
||||
await agentsSection.screenshot({
|
||||
path: `${SHOTS}/03-mixed-states.png`,
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user