Skip to content

Commit 190f51c

Browse files
recuu-pfegrecuu-pfeg
committed
refactor: extract PollLoop, deduplicate polling in Manager and MessagePoller
- Create src/poll-loop.ts: reusable PollLoop with guard, pause/resume, error handling - Refactor Manager to use PollLoop (removes manual timer/processing fields) - Refactor MessagePoller to use PollLoop - ChatPoller left as-is (deprecated)
1 parent 6560921 commit 190f51c

3 files changed

Lines changed: 121 additions & 64 deletions

File tree

‎src/manager.ts‎

Lines changed: 43 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import type { ApiClient, Task } from "./api-client.js";
1616
import type { AgentRunner } from "./runner.js";
1717
import { callClaudeCli, createAuthHeaders, buildConversationHistory } from "./llm-client.js";
18+
import { PollLoop } from "./poll-loop.js";
1819
import * as ui from "./ui.js";
1920

2021
// ---------------------------------------------------------------------------
@@ -94,8 +95,7 @@ export class Manager {
9495
private runner: AgentRunner | null;
9596
private api: ApiClient | null;
9697
private lastSeenId: string | null = null;
97-
private timer: ReturnType<typeof setInterval> | null = null;
98-
private processing = false;
98+
private poller: PollLoop;
9999
private useClaudeCli: boolean;
100100

101101
/** Callbacks for actions the Manager can't execute directly */
@@ -115,39 +115,35 @@ export class Manager {
115115
this.runner = options.runner ?? null;
116116
this.api = options.api ?? null;
117117
this.useClaudeCli = !options.llmBaseUrl || !options.llmApiKey;
118+
this.poller = new PollLoop({
119+
name: "manager",
120+
intervalMs: this.pollIntervalMs,
121+
onTick: () => this.poll(),
122+
});
118123
}
119124

120125
// ── Lifecycle ────────────────────────────────────────────
121126

122127
start(): void {
123128
ui.info(`[manager] Started (poll every ${this.pollIntervalMs / 1000}s, model: ${this.model})`);
124-
this.timer = setInterval(() => this.poll(), this.pollIntervalMs);
125-
this.poll();
129+
this.poller.start();
126130
}
127131

128132
stop(): void {
129-
if (this.timer) {
130-
clearInterval(this.timer);
131-
this.timer = null;
132-
}
133+
this.poller.stop();
133134
ui.debug("manager", "Manager stopped");
134135
}
135136

136137
/** Pause polling when WS clients are connected (messages come via WS) */
137138
pausePolling(): void {
138-
if (this.timer) {
139-
clearInterval(this.timer);
140-
this.timer = null;
141-
ui.info("[manager] Polling paused (WS connected)");
142-
}
139+
ui.info("[manager] Polling paused (WS connected)");
140+
this.poller.pause();
143141
}
144142

145143
/** Resume polling when all WS clients disconnected */
146144
resumePolling(): void {
147-
if (!this.timer) {
148-
this.timer = setInterval(() => this.poll(), this.pollIntervalMs);
149-
ui.info("[manager] Polling resumed (no WS clients)");
150-
}
145+
ui.info("[manager] Polling resumed (no WS clients)");
146+
this.poller.resume();
151147
}
152148

153149
// ── WebSocket entry point ────────────────────────────────
@@ -183,52 +179,43 @@ export class Manager {
183179
// ── Polling ──────────────────────────────────────────────
184180

185181
private async poll(): Promise<void> {
186-
if (this.processing) return;
187-
this.processing = true;
182+
await this.heartbeat();
188183

189-
try {
190-
await this.heartbeat();
191-
192-
const messages = await this.fetchMessages();
193-
if (!messages || messages.length === 0) return;
184+
const messages = await this.fetchMessages();
185+
if (!messages || messages.length === 0) return;
194186

195-
const newMessages = this.findNewInboundMessages(messages);
196-
if (newMessages.length > 0) {
197-
ui.debug("manager", `${newMessages.length} new message(s) to process`);
198-
}
187+
const newMessages = this.findNewInboundMessages(messages);
188+
if (newMessages.length > 0) {
189+
ui.debug("manager", `${newMessages.length} new message(s) to process`);
190+
}
199191

200-
for (const msg of newMessages) {
201-
const isFromUser = msg.from === "user" || msg.from.startsWith("user:");
192+
for (const msg of newMessages) {
193+
const isFromUser = msg.from === "user" || msg.from.startsWith("user:");
202194

203-
try {
204-
const senderContext = isFromUser
205-
? `[Message from user: ${msg.from}]`
206-
: `[Message from agent: ${msg.from}]`;
207-
const enrichedContent = `${senderContext}\n${msg.content}`;
208-
209-
const context = await this.fetchContext();
210-
const { reply, actions, proposals } = await this.think(enrichedContent, context);
211-
await this.executeActions(actions, context);
212-
await this.postReplyTo(msg.from, reply);
213-
if (isFromUser) {
214-
this.onReply?.(reply);
215-
if (proposals && proposals.length > 0) {
216-
this.onProposals?.(proposals);
217-
}
195+
try {
196+
const senderContext = isFromUser
197+
? `[Message from user: ${msg.from}]`
198+
: `[Message from agent: ${msg.from}]`;
199+
const enrichedContent = `${senderContext}\n${msg.content}`;
200+
201+
const context = await this.fetchContext();
202+
const { reply, actions, proposals } = await this.think(enrichedContent, context);
203+
await this.executeActions(actions, context);
204+
await this.postReplyTo(msg.from, reply);
205+
if (isFromUser) {
206+
this.onReply?.(reply);
207+
if (proposals && proposals.length > 0) {
208+
this.onProposals?.(proposals);
218209
}
219-
// Compact log: inbound + reply on two lines
220-
ui.chatExchange(msg.from, msg.content, reply, actions.length);
221-
} catch (err) {
222-
ui.chatExchange(msg.from, msg.content, `Error: ${err}`, 0);
223-
await this.postReplyTo(msg.from, "Sorry, an error occurred while processing your message. Please try again.").catch(() => {});
224210
}
225-
226-
this.lastSeenId = msg.id;
211+
// Compact log: inbound + reply on two lines
212+
ui.chatExchange(msg.from, msg.content, reply, actions.length);
213+
} catch (err) {
214+
ui.chatExchange(msg.from, msg.content, `Error: ${err}`, 0);
215+
await this.postReplyTo(msg.from, "Sorry, an error occurred while processing your message. Please try again.").catch(() => {});
227216
}
228-
} catch (err) {
229-
ui.debug("manager", `Poll error: ${err}`);
230-
} finally {
231-
this.processing = false;
217+
218+
this.lastSeenId = msg.id;
232219
}
233220
}
234221

‎src/message-poller.ts‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import { writeFileSync, existsSync, readFileSync, statSync } from "node:fs";
1010
import { join } from "node:path";
1111
import type { ApiClient, Message } from "./api-client.js";
12+
import { PollLoop } from "./poll-loop.js";
1213
import * as ui from "./ui.js";
1314

1415
const POLL_INTERVAL_MS = 10_000;
@@ -28,34 +29,34 @@ export class MessagePoller {
2829
private api: ApiClient;
2930
private channel: string;
3031
private workingDir: string;
31-
private timer: ReturnType<typeof setInterval> | null = null;
32+
private poller: PollLoop;
3233
private lastSeenTimestamp: string | null = null;
3334
private deliveredIds = new Set<string>();
3435

3536
constructor(options: MessagePollerOptions) {
3637
this.api = options.api;
3738
this.channel = options.channel;
3839
this.workingDir = options.workingDir;
40+
this.poller = new PollLoop({
41+
name: `msg:${this.channel}`,
42+
intervalMs: POLL_INTERVAL_MS,
43+
onTick: () => this.poll(),
44+
});
3945
}
4046

4147
/**
4248
* Start polling for messages.
4349
*/
4450
start(): void {
4551
ui.debug("msg", `Polling messages for "${this.channel}" every ${POLL_INTERVAL_MS / 1000}s`);
46-
// Run immediately, then on interval
47-
this.poll();
48-
this.timer = setInterval(() => this.poll(), POLL_INTERVAL_MS);
52+
this.poller.start();
4953
}
5054

5155
/**
5256
* Stop polling and clean up.
5357
*/
5458
stop(): void {
55-
if (this.timer) {
56-
clearInterval(this.timer);
57-
this.timer = null;
58-
}
59+
this.poller.stop();
5960
}
6061

6162
private async poll(): Promise<void> {

‎src/poll-loop.ts‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
/**
2+
* Reusable poll loop with guard, pause/resume, and error handling.
3+
*/
4+
import * as ui from "./ui.js";
5+
6+
export interface PollLoopOptions {
7+
name: string;
8+
intervalMs: number;
9+
onTick: () => Promise<void>;
10+
onError?: (err: unknown) => void;
11+
}
12+
13+
export class PollLoop {
14+
private timer: ReturnType<typeof setInterval> | null = null;
15+
private processing = false;
16+
private readonly name: string;
17+
private readonly intervalMs: number;
18+
private readonly onTick: () => Promise<void>;
19+
private readonly onError: (err: unknown) => void;
20+
21+
constructor(opts: PollLoopOptions) {
22+
this.name = opts.name;
23+
this.intervalMs = opts.intervalMs;
24+
this.onTick = opts.onTick;
25+
this.onError = opts.onError ?? ((err) => ui.warn(`[${this.name}] poll error: ${err}`));
26+
}
27+
28+
start(): void {
29+
if (this.timer) return;
30+
this.timer = setInterval(() => this.poll(), this.intervalMs);
31+
this.poll(); // Run immediately
32+
}
33+
34+
stop(): void {
35+
if (this.timer) {
36+
clearInterval(this.timer);
37+
this.timer = null;
38+
}
39+
}
40+
41+
pause(): void {
42+
if (this.timer) {
43+
clearInterval(this.timer);
44+
this.timer = null;
45+
}
46+
}
47+
48+
resume(): void {
49+
if (!this.timer) {
50+
this.timer = setInterval(() => this.poll(), this.intervalMs);
51+
}
52+
}
53+
54+
get isRunning(): boolean {
55+
return this.timer !== null;
56+
}
57+
58+
private async poll(): Promise<void> {
59+
if (this.processing) return;
60+
this.processing = true;
61+
try {
62+
await this.onTick();
63+
} catch (err) {
64+
this.onError(err);
65+
} finally {
66+
this.processing = false;
67+
}
68+
}
69+
}

0 commit comments

Comments
 (0)