mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-24 18:22:38 +00:00
fix daemon stale-progress semantics and hydrate replay
This commit is contained in:
@@ -49,7 +49,11 @@ import { derivePendingPermissionKey, normalizeAgentSnapshot } from "@/utils/agen
|
||||
import { resolveProjectPlacement } from "@/utils/project-placement";
|
||||
import { buildDraftStoreKey } from "@/stores/draft-keys";
|
||||
import type { AttachmentMetadata } from "@/attachments/types";
|
||||
import { takeQueuedAgentMessageReplay } from "@/contexts/session-queued-message-replay";
|
||||
import {
|
||||
shouldAutoReplayQueuedAgentMessage,
|
||||
takeQueuedAgentMessageReplay,
|
||||
type QueuedAgentReplaySource,
|
||||
} from "@/contexts/session-queued-message-replay";
|
||||
|
||||
// Re-export types from session-store and draft-store for backward compatibility
|
||||
export type { DraftInput } from "@/stores/draft-store";
|
||||
@@ -355,7 +359,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
);
|
||||
|
||||
const applyAuthoritativeAgentSnapshot = useCallback(
|
||||
(agent: Agent) => {
|
||||
(agent: Agent, options?: { source?: QueuedAgentReplaySource }) => {
|
||||
setAgents(serverId, (prev) => {
|
||||
const current = prev.get(agent.id);
|
||||
if (current && agent.updatedAt.getTime() < current.updatedAt.getTime()) {
|
||||
@@ -429,7 +433,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
});
|
||||
|
||||
const prevStatus = previousAgentStatusRef.current.get(agent.id);
|
||||
if (prevStatus === "running" && agent.status !== "running") {
|
||||
if (
|
||||
shouldAutoReplayQueuedAgentMessage({
|
||||
previousStatus: prevStatus,
|
||||
nextStatus: agent.status,
|
||||
source: options?.source ?? "live",
|
||||
})
|
||||
) {
|
||||
const session = useSessionStore.getState().sessions[serverId];
|
||||
const replay = takeQueuedAgentMessageReplay(session?.queuedMessages.get(agent.id));
|
||||
if (replay) {
|
||||
@@ -732,7 +742,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
}),
|
||||
};
|
||||
|
||||
applyAuthoritativeAgentSnapshot(agent);
|
||||
applyAuthoritativeAgentSnapshot(agent, { source: "live" });
|
||||
},
|
||||
[
|
||||
applyAuthoritativeAgentSnapshot,
|
||||
@@ -782,10 +792,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
|
||||
if (payload.agent) {
|
||||
const normalized = normalizeAgentSnapshot(payload.agent, serverId);
|
||||
applyAuthoritativeAgentSnapshot({
|
||||
...normalized,
|
||||
projectPlacement: session?.agents.get(agentId)?.projectPlacement ?? null,
|
||||
});
|
||||
applyAuthoritativeAgentSnapshot(
|
||||
{
|
||||
...normalized,
|
||||
projectPlacement: session?.agents.get(agentId)?.projectPlacement ?? null,
|
||||
},
|
||||
{ source: "hydrate" },
|
||||
);
|
||||
}
|
||||
|
||||
// Call pure reducer
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { takeQueuedAgentMessageReplay } from "./session-queued-message-replay";
|
||||
import {
|
||||
shouldAutoReplayQueuedAgentMessage,
|
||||
takeQueuedAgentMessageReplay,
|
||||
} from "./session-queued-message-replay";
|
||||
|
||||
describe("takeQueuedAgentMessageReplay", () => {
|
||||
it("preserves the queued message id for idempotent replay", () => {
|
||||
@@ -47,4 +50,30 @@ describe("takeQueuedAgentMessageReplay", () => {
|
||||
expect(takeQueuedAgentMessageReplay([])).toBeNull();
|
||||
expect(takeQueuedAgentMessageReplay(undefined)).toBeNull();
|
||||
});
|
||||
|
||||
it("only auto-replays queued messages for live running-to-idle transitions", () => {
|
||||
expect(
|
||||
shouldAutoReplayQueuedAgentMessage({
|
||||
previousStatus: "running",
|
||||
nextStatus: "idle",
|
||||
source: "live",
|
||||
}),
|
||||
).toBe(true);
|
||||
|
||||
expect(
|
||||
shouldAutoReplayQueuedAgentMessage({
|
||||
previousStatus: "running",
|
||||
nextStatus: "idle",
|
||||
source: "hydrate",
|
||||
}),
|
||||
).toBe(false);
|
||||
|
||||
expect(
|
||||
shouldAutoReplayQueuedAgentMessage({
|
||||
previousStatus: "idle",
|
||||
nextStatus: "idle",
|
||||
source: "live",
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -13,6 +13,16 @@ export interface QueuedAgentMessageReplay {
|
||||
remainingQueue: QueuedAgentMessageReplayItem[];
|
||||
}
|
||||
|
||||
export type QueuedAgentReplaySource = "hydrate" | "live";
|
||||
|
||||
export function shouldAutoReplayQueuedAgentMessage(input: {
|
||||
previousStatus: string | undefined;
|
||||
nextStatus: string;
|
||||
source: QueuedAgentReplaySource;
|
||||
}): boolean {
|
||||
return input.source === "live" && input.previousStatus === "running" && input.nextStatus !== "running";
|
||||
}
|
||||
|
||||
export function takeQueuedAgentMessageReplay(
|
||||
queue: readonly QueuedAgentMessageReplayItem[] | undefined,
|
||||
): QueuedAgentMessageReplay | null {
|
||||
|
||||
@@ -1588,6 +1588,34 @@ describe("AgentManager", () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("notifyAgentState does not mutate updatedAt", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
const storage = new AgentStorage(storagePath, logger);
|
||||
const manager = new AgentManager({
|
||||
clients: {
|
||||
codex: new TestAgentClient(),
|
||||
},
|
||||
registry: storage,
|
||||
logger,
|
||||
idFactory: () => "00000000-0000-4000-8000-000000000121",
|
||||
});
|
||||
|
||||
const snapshot = await manager.createAgent({
|
||||
provider: "codex",
|
||||
cwd: workdir,
|
||||
});
|
||||
|
||||
const before = manager.getAgent(snapshot.id);
|
||||
expect(before).toBeDefined();
|
||||
|
||||
manager.notifyAgentState(snapshot.id);
|
||||
|
||||
const after = manager.getAgent(snapshot.id);
|
||||
expect(after).toBeDefined();
|
||||
expect(after!.updatedAt.getTime()).toBe(before!.updatedAt.getTime());
|
||||
});
|
||||
|
||||
test("recordUserMessage can skip emitting agent_state when run start will emit running", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
|
||||
@@ -847,7 +847,6 @@ export class AgentManager {
|
||||
if (!agent || agent.internal) {
|
||||
return;
|
||||
}
|
||||
this.touchUpdatedAt(agent);
|
||||
this.emitState(agent);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user