Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- **The codex runtime's review-trigger loop reads the agent's own catch-up feed (Closes #631).** It uses a different cursor filename from the task loop and accepts loop IDs through 79 characters, including a 64-character agent ID plus `-review-trigger`.
13 changes: 3 additions & 10 deletions packages/cli/src/utils/codex-runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import {
} from "./agent-lifecycle.js";
import { FlairClient, type PresenceActivity } from "./flair-client.js";
import { startTaskLoop } from "./flair-task-loop.js";
import { handlePrOpened } from "./pr-review-trigger.js";
import { startReviewTriggerLoop } from "./pr-review-trigger.js";
import { completeRuntimeMail, pollRuntimeMail, type RuntimeMailConfig, runtimeBootPreflight, sendRuntimeMail } from "./runtime-mail.js";
import { externalDispatchRefusal } from "./mail-tier.js";
import type { Envelope } from "@tpsdev-ai/agent";
Expand Down Expand Up @@ -853,16 +853,9 @@ export async function runCodexRuntime(config: CodexRuntimeConfig): Promise<void>
}
});

// Review trigger — listens for pr.opened and auto-requests K&S review
// Review trigger — polls the agent's own feed using a different cursor filename from the task loop.
if (config.reviewTrigger) {
const triggerConfig = {
reviewers: config.reviewTrigger.reviewers,
agentId,
repo: config.reviewTrigger.repo,
};
startTaskLoop(flair, `${agentId}-review-trigger`, async (event) => {
await handlePrOpened(event, triggerConfig);
}, { kinds: ["pr.opened"] });
startReviewTriggerLoop(flair, agentId, config.reviewTrigger);
console.log(`[${agentId}] Review trigger active → ${config.reviewTrigger.reviewers.join(", ")}`);
}

Expand Down
75 changes: 53 additions & 22 deletions packages/cli/src/utils/flair-task-loop.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,16 @@
* flair-task-loop.ts — Polls OrgEventCatchup for task.assigned events
* targeting this agent. Runtime-agnostic: works without OpenClaw.
*
* CURSOR. The loop reads the catch-up feed through `FlairClient.getEventsSince`
* (the validated reader in org-event-read.ts, cli#619), always sending its
* position as Flair's `after` cursor, so Flair's own watermark for the agent
* never chooses where reading starts. The position is persisted in
* `~/.tps/cursors/<agentId>-task-loop.json` as `{ position, updatedAt }` and
* replaced atomically (a temporary file in the same directory, fsynced, then
* renamed over it). With no usable cursor file, the loop sets the position to
* CURSOR. The loop reads `agentId`'s catch-up feed through
* `FlairClient.getEventsSince` (the validated reader in org-event-read.ts,
* cli#619), always sending its position as Flair's `after` cursor, so Flair's
* own watermark for the agent never chooses where reading starts. The position
* is persisted in `~/.tps/cursors/<loopId>-task-loop.json` as
* `{ position, updatedAt }` and replaced atomically (a temporary file in the
* same directory, fsynced, then renamed over it). The task and review loops use
* different cursor filenames; callers that omit `loopId` use the `agentId` cursor. It must
* be a single safe filename component (see `LOOP_ID_PATTERN`); `startTaskLoop` refuses any other with
* `InvalidLoopIdError` before reading or writing a cursor. With no usable cursor file, the loop sets the position to
* FIRST_BOOT_LOOKBACK_MS before now, once, and does not read the feed until
* that position is saved: a failed save is logged by name and retried with
* backoff, with the same position, and `stop()` ends the retrying. A legacy
Expand Down Expand Up @@ -95,9 +98,31 @@ function cursorDir(): string {
return join(homeDir(), ".tps", "cursors");
}

/** The task loop's cursor file for `agentId`. */
export function taskLoopCursorPath(agentId: string): string {
return join(cursorDir(), `${agentId}-task-loop.json`);
/**
* The allowed set for `loopId`: a single filename component of 1..79 characters
* from `A-Z a-z 0-9 . _ -`, excluding the two dot components `.` and `..`. It
* names the cursor file `<loopId>-task-loop.json` inside the cursors directory,
* so a path separator, a NUL, an empty or over-long string, or `.`/`..` is
* refused rather than allowed to resolve a cursor outside that directory.
*/
const LOOP_ID_PATTERN = /^[A-Za-z0-9._-]{1,79}$/;

/** Whether `loopId` is a single safe filename component (see `LOOP_ID_PATTERN`). */
export function isSafeLoopId(loopId: string): boolean {
return LOOP_ID_PATTERN.test(loopId) && loopId !== "." && loopId !== "..";
}

/** Thrown by `startTaskLoop` for a `loopId` that is not a single safe filename component. */
export class InvalidLoopIdError extends Error {
constructor(loopId: string) {
super(`Invalid loopId ${JSON.stringify(loopId)}: it must be a single filename component matching ${LOOP_ID_PATTERN} and not "." or ".."`);
this.name = "InvalidLoopIdError";
}
}

/** The loop's cursor file for `loopId`. */
export function taskLoopCursorPath(loopId: string): string {
return join(cursorDir(), `${loopId}-task-loop.json`);
}

/**
Expand Down Expand Up @@ -125,10 +150,10 @@ export function cursorFilePosition(data: unknown): string | undefined {
}

/** The persisted feed position (Flair's `after` cursor), or undefined when there is none to resume from. */
function loadCursor(agentId: string): string | undefined {
const path = taskLoopCursorPath(agentId);
function loadCursor(loopId: string): string | undefined {
const path = taskLoopCursorPath(loopId);
if (!existsSync(path)) {
slog(`[${agentId}] No task loop cursor at ${path}; ${LOOKBACK_NOTE}`);
slog(`[${loopId}] No task loop cursor at ${path}; ${LOOKBACK_NOTE}`);
return undefined;
}
let position: string | undefined;
Expand All @@ -139,7 +164,7 @@ function loadCursor(agentId: string): string | undefined {
problem = e instanceof SyntaxError ? "it is not JSON" : `it cannot be read (${(e as NodeJS.ErrnoException).code ?? (e as Error).name})`;
}
if (position === undefined) {
swarn(`[${agentId}] Task loop cursor at ${path} is unusable (${problem}); treating it as no cursor, so ${LOOKBACK_NOTE}`);
swarn(`[${loopId}] Task loop cursor at ${path} is unusable (${problem}); treating it as no cursor, so ${LOOKBACK_NOTE}`);
}
return position;
}
Expand All @@ -149,8 +174,8 @@ function loadCursor(agentId: string): string | undefined {
* it, rename it over the cursor. Returns undefined on success, or the failure's
* name (its code, such as EACCES, or its class) for the caller to log.
*/
function saveCursor(agentId: string, position: string): string | undefined {
const path = taskLoopCursorPath(agentId);
function saveCursor(loopId: string, position: string): string | undefined {
const path = taskLoopCursorPath(loopId);
const tmp = `${path}.${process.pid}.${Date.now()}.tmp`;
try {
mkdirSync(cursorDir(), { recursive: true });
Expand Down Expand Up @@ -190,11 +215,17 @@ export function startTaskLoop(
flair: FlairClient,
agentId: string,
handler: TaskHandler,
opts: { pollIntervalMs?: number; kinds?: string[] } = {},
opts: { pollIntervalMs?: number; kinds?: string[]; loopId?: string } = {},
): { stop: () => Promise<void> } {
const basePollMs = opts.pollIntervalMs ?? DEFAULT_POLL_MS;
const kinds = new Set(opts.kinds ?? ["task.assigned"]);
const loaded = loadCursor(agentId);
// Omitted loop IDs use the agentId cursor.
const loopId = opts.loopId ?? agentId;
// Refuse an unsafe loopId before any cursor is read or written: it names the
// cursor file, so one with a path separator or a traversal segment could
// resolve a cursor outside the cursors directory.
if (!isSafeLoopId(loopId)) throw new InvalidLoopIdError(loopId);
const loaded = loadCursor(loopId);
// A new initial position is computed once, here; saving it is retried with
// this same value until it succeeds, and the feed is not read before then.
let position: string = loaded ?? initialPosition();
Expand All @@ -216,11 +247,11 @@ export function startTaskLoop(
/** Save the initial position, retrying with backoff; false when stopped first. */
async function saveInitialPosition(): Promise<boolean> {
for (let attempt = 1; running; attempt++) {
const failure = saveCursor(agentId, position);
const failure = saveCursor(loopId, position);
if (failure === undefined) return true;
const delay = Math.min(basePollMs * 2 ** (attempt - 1), MAX_BACKOFF_MS);
swarn(
`[${agentId}] Task loop cannot save its initial position to ${taskLoopCursorPath(agentId)} (${failure}); ` +
`[${loopId}] Task loop cannot save its initial position to ${taskLoopCursorPath(loopId)} (${failure}); ` +
`the feed is not read until it is saved; retrying in ${delay} ms`,
);
await sleep(delay);
Expand Down Expand Up @@ -290,8 +321,8 @@ export function startTaskLoop(
// A failed save is logged and the loop goes on from the position in
// memory. The saved position is then behind it, so a restart replays
// events rather than skipping them.
const failure = saveCursor(agentId, position);
if (failure !== undefined) swarn(`[${agentId}] Failed to persist cursor to ${taskLoopCursorPath(agentId)} (${failure})`);
const failure = saveCursor(loopId, position);
if (failure !== undefined) swarn(`[${loopId}] Failed to persist cursor to ${taskLoopCursorPath(loopId)} (${failure})`);
}

// Cap seen set
Expand Down
24 changes: 23 additions & 1 deletion packages/cli/src/utils/pr-review-trigger.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@

import { spawnSync } from "node:child_process";
import snooplogg from "snooplogg";
import type { FlairClient } from "./flair-client.js";
import { startTaskLoop } from "./flair-task-loop.js";
const { log: slog, warn: swarn, error: serror } = snooplogg("tps:github");


Expand Down Expand Up @@ -74,6 +76,7 @@ export function reRequestReviewer(
export async function handlePrOpened(
event: { detail?: string; summary: string; refId?: string },
config: ReviewTriggerConfig,
deps: ReviewRequestDeps = {},
): Promise<void> {
const detail = event.detail ?? "";
const prNumber = extractPrNumber(detail);
Expand All @@ -83,5 +86,24 @@ export async function handlePrOpened(
return;
}

requestReviews(prNumber, config);
requestReviews(prNumber, config, deps);
}

/**
* Poll the agent's own feed for `pr.opened`, using a different cursor filename from the task loop.
* The loop ID allows the CLI's 64-character agent ID plus `-review-trigger`.
*/
export function startReviewTriggerLoop(
flair: FlairClient,
agentId: string,
config: { reviewers: string[]; repo: string },
opts: { pollIntervalMs?: number; deps?: ReviewRequestDeps } = {},
): { stop: () => Promise<void> } {
const triggerConfig: ReviewTriggerConfig = { reviewers: config.reviewers, agentId, repo: config.repo };
return startTaskLoop(
flair,
agentId,
async (event) => { await handlePrOpened(event, triggerConfig, opts.deps); },
{ kinds: ["pr.opened"], loopId: `${agentId}-review-trigger`, pollIntervalMs: opts.pollIntervalMs },
);
}
46 changes: 35 additions & 11 deletions packages/cli/test/flair-task-loop.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,13 @@
* the signing and parsing seams are exercised; only the HTTP endpoint is faked.
*/
import { afterEach, beforeEach, describe, expect, test } from "bun:test";
import { chmodSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs";
import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { Writable } from "node:stream";
import { SnoopLogg } from "snooplogg";
import { createFlairClient } from "../src/utils/flair-client.js";
import { MAX_POSITION_SKEW_MS, startTaskLoop } from "../src/utils/flair-task-loop.js";
import { createFlairClient, type FlairClient } from "../src/utils/flair-client.js";
import { InvalidLoopIdError, MAX_POSITION_SKEW_MS, startTaskLoop } from "../src/utils/flair-task-loop.js";
import { createPatchShared } from "./helpers/patch-shared.js";
import {
type CatchupFeed,
Expand Down Expand Up @@ -507,16 +507,16 @@ describe("cli#624 — the initial position is set once and always sent", () => {
});

describe("cli#624 — the initial position is saved before the first read", () => {
// A real write failure: the cursor directory is made unwritable, then restored.
// A regular file blocks creation of the cursor directory, including for root.
const cursors = () => join(home, ".tps", "cursors");
const lockCursors = () => {
mkdirSync(cursors(), { recursive: true });
chmodSync(cursors(), 0o500);
mkdirSync(join(home, ".tps"), { recursive: true });
writeFileSync(cursors(), "blocked");
};
const unlockCursors = () => chmodSync(cursors(), 0o700);
const unlockCursors = () => rmSync(cursors());
const saveFailures = () => warnings.split("cannot save its initial position").length - 1;

test.skipIf(process.getuid?.() === 0)(
test(
"while the initial position cannot be saved nothing is read; after recovery the saved position is the one fixed at start, and a restart retries the first task",
async () => {
const startNow = Date.now();
Expand All @@ -535,9 +535,9 @@ describe("cli#624 — the initial position is saved before the first read", () =
}, { pollIntervalMs: 20 });
try {
expect(await waitFor(() => saveFailures() >= 3)).toBe(true);
expect(warnings).toContain(`cannot save its initial position to ${cursorFile()} (EACCES)`);
expect(warnings).toContain(`cannot save its initial position to ${cursorFile()} (EEXIST)`);
expect(catchup.afters).toEqual([]);
// Half an hour passes before the directory is writable again: an initial
// Half an hour passes before the directory can be created: an initial
// position recomputed now would start after e1.
offset = 30 * 60_000;
unlockCursors();
Expand Down Expand Up @@ -567,7 +567,7 @@ describe("cli#624 — the initial position is saved before the first read", () =
},
);

test.skipIf(process.getuid?.() === 0)("stop ends the retrying of the initial save promptly, and nothing is read", async () => {
test("stop ends the retrying of the initial save promptly, and nothing is read", async () => {
const catchup = serveFeed({ events: [assigned("e1", 5)] });
lockCursors();
try {
Expand Down Expand Up @@ -709,3 +709,27 @@ describe("cli#624 — the task loop's cursor file", () => {
expect(readdirSync(join(home, ".tps", "cursors"))).toEqual([`${AGENT}-task-loop.json`]);
});
});

describe("cli#631 — loopId is a single safe filename component", () => {
test("refuses a traversal, separator, dot or over-long loopId before reading or writing a cursor", () => {
// A client the loop could not use even if it got past the check: the refusal must come first.
const untouched = {} as FlairClient;
for (const bad of ["../../outside", "a/b", "..", ".", "a\\b", "x".repeat(80), "a\0b"]) {
expect(() => startTaskLoop(untouched, AGENT, async () => {}, { loopId: bad })).toThrow(InvalidLoopIdError);
}
// Nothing was read or written: the cursors directory does not exist, and the
// path a traversal loopId would reach is untouched.
expect(existsSync(join(home, ".tps", "cursors"))).toBe(false);
expect(existsSync(join(home, "outside-task-loop.json"))).toBe(false);
});

test("accepts a valid loopId and names its cursor file with it", async () => {
serveFeed({ events: [] });
const loop = startTaskLoop(client(), AGENT, async () => {}, { loopId: "agent-a-review-trigger", pollIntervalMs: 60_000 });
try {
expect(await waitFor(() => existsSync(join(home, ".tps", "cursors", "agent-a-review-trigger-task-loop.json")))).toBe(true);
} finally {
await loop.stop();
}
});
});
Loading
Loading