Code Queue Review
322 lines
11 KiB
JavaScript
322 lines
11 KiB
JavaScript
import { randomUUID } from "node:crypto";
|
|
|
|
export const DEFAULT_CODE_AGENT_SESSION_IDLE_TIMEOUT_MS = 30 * 60 * 1000;
|
|
export const DEFAULT_CODE_AGENT_MAX_SESSIONS = 200;
|
|
export const CODE_AGENT_SESSION_STATUSES = Object.freeze([
|
|
"creating",
|
|
"ready",
|
|
"busy",
|
|
"idle",
|
|
"interrupted",
|
|
"expired",
|
|
"failed"
|
|
]);
|
|
|
|
export function createCodeAgentSessionRegistry(options = {}) {
|
|
const idleTimeoutMs = positiveInteger(options.idleTimeoutMs, DEFAULT_CODE_AGENT_SESSION_IDLE_TIMEOUT_MS);
|
|
const maxSessions = positiveInteger(options.maxSessions, DEFAULT_CODE_AGENT_MAX_SESSIONS);
|
|
const idFactory = typeof options.idFactory === "function" ? options.idFactory : () => `ses_${randomUUID()}`;
|
|
const defaultNow = options.now;
|
|
const sessions = new Map();
|
|
const conversations = new Map();
|
|
|
|
function acquire(params = {}) {
|
|
const timestamp = timestampFor(params.now ?? defaultNow);
|
|
const timestampMs = Date.parse(timestamp);
|
|
const conversationId = requiredId(params.conversationId, "cnv");
|
|
const requestedSessionId = optionalId(params.sessionId);
|
|
const mappedSessionId = conversations.get(conversationId) ?? null;
|
|
|
|
if (mappedSessionId && requestedSessionId && mappedSessionId !== requestedSessionId) {
|
|
const mappedSession = sessions.get(mappedSessionId) ?? null;
|
|
return blockedAcquire({
|
|
code: "session_reuse_conflict",
|
|
summary: `conversationId ${conversationId} is already bound to sessionId ${mappedSessionId}; requested ${requestedSessionId}.`,
|
|
session: mappedSession,
|
|
timestamp,
|
|
traceId: params.traceId
|
|
});
|
|
}
|
|
|
|
const effectiveSessionId = mappedSessionId || requestedSessionId || requiredId(idFactory(), "ses");
|
|
let session = sessions.get(effectiveSessionId) ?? null;
|
|
const reused = Boolean(session);
|
|
|
|
if (session && sessionExpired(session, timestampMs)) {
|
|
session.status = "expired";
|
|
session.updatedAt = timestamp;
|
|
session.lastTraceId = optionalId(params.traceId);
|
|
return blockedAcquire({
|
|
code: "session_expired",
|
|
summary: `Code Agent runner session ${effectiveSessionId} expired after ${session.idleTimeoutMs}ms idle timeout.`,
|
|
session,
|
|
timestamp,
|
|
traceId: params.traceId
|
|
});
|
|
}
|
|
|
|
if (session?.status === "busy") {
|
|
session.updatedAt = timestamp;
|
|
session.lastTraceId = optionalId(params.traceId) ?? session.lastTraceId;
|
|
return blockedAcquire({
|
|
code: "session_busy",
|
|
summary: `Code Agent runner session ${effectiveSessionId} is already busy with traceId ${session.currentTraceId ?? "unknown"}.`,
|
|
session,
|
|
timestamp,
|
|
traceId: params.traceId
|
|
});
|
|
}
|
|
|
|
if (session && ["failed", "interrupted"].includes(session.status)) {
|
|
session.updatedAt = timestamp;
|
|
session.lastTraceId = optionalId(params.traceId) ?? session.lastTraceId;
|
|
return blockedAcquire({
|
|
code: `session_${session.status}`,
|
|
summary: `Code Agent runner session ${effectiveSessionId} is ${session.status}; create a new session before retrying.`,
|
|
session,
|
|
timestamp,
|
|
traceId: params.traceId
|
|
});
|
|
}
|
|
|
|
if (!session) {
|
|
session = {
|
|
sessionId: effectiveSessionId,
|
|
conversationIds: new Set(),
|
|
status: "creating",
|
|
workspace: params.workspace ?? null,
|
|
sandbox: params.sandbox ?? "read-only",
|
|
runnerKind: params.runnerKind ?? "hwlab-readonly-runner",
|
|
sessionMode: params.sessionMode ?? params.implementationType ?? "controlled-readonly-session-registry",
|
|
capabilityLevel: params.capabilityLevel ?? "read-only-session-tools",
|
|
implementationType: params.implementationType ?? "controlled-readonly-session-registry",
|
|
createdAt: timestamp,
|
|
updatedAt: timestamp,
|
|
idleTimeoutMs,
|
|
expiresAt: plusMs(timestampMs, idleTimeoutMs),
|
|
lastTraceId: optionalId(params.traceId),
|
|
currentTraceId: null,
|
|
turn: 0,
|
|
durable: false,
|
|
longLivedSession: true,
|
|
codexStdio: false,
|
|
readOnly: true,
|
|
writeCapable: false,
|
|
secretMaterialStored: false
|
|
};
|
|
sessions.set(effectiveSessionId, session);
|
|
}
|
|
|
|
session.conversationIds.add(conversationId);
|
|
session.workspace = params.workspace ?? session.workspace;
|
|
session.sandbox = params.sandbox ?? session.sandbox;
|
|
session.runnerKind = params.runnerKind ?? session.runnerKind;
|
|
session.sessionMode = params.sessionMode ?? session.sessionMode;
|
|
session.capabilityLevel = params.capabilityLevel ?? session.capabilityLevel;
|
|
session.implementationType = params.implementationType ?? session.implementationType;
|
|
session.status = "busy";
|
|
session.updatedAt = timestamp;
|
|
session.expiresAt = plusMs(timestampMs, idleTimeoutMs);
|
|
session.lastTraceId = optionalId(params.traceId);
|
|
session.currentTraceId = optionalId(params.traceId);
|
|
session.turn += 1;
|
|
conversations.set(conversationId, effectiveSessionId);
|
|
prune();
|
|
|
|
return {
|
|
ok: true,
|
|
reused,
|
|
session: publicSession(session, { conversationId, reused, timestamp })
|
|
};
|
|
}
|
|
|
|
function release(sessionId, params = {}) {
|
|
const session = sessions.get(requiredId(sessionId, "ses")) ?? null;
|
|
if (!session) return null;
|
|
const timestamp = timestampFor(params.now ?? defaultNow);
|
|
const timestampMs = Date.parse(timestamp);
|
|
session.status = CODE_AGENT_SESSION_STATUSES.includes(params.status) ? params.status : "idle";
|
|
session.updatedAt = timestamp;
|
|
session.expiresAt = plusMs(timestampMs, session.idleTimeoutMs);
|
|
session.lastTraceId = optionalId(params.traceId) ?? session.lastTraceId;
|
|
session.currentTraceId = null;
|
|
session.statusReason = params.statusReason ?? null;
|
|
return publicSession(session, {
|
|
conversationId: params.conversationId,
|
|
reused: params.reused,
|
|
timestamp
|
|
});
|
|
}
|
|
|
|
function fail(sessionId, params = {}) {
|
|
return release(sessionId, { ...params, status: "failed" });
|
|
}
|
|
|
|
function interrupt(sessionId, params = {}) {
|
|
return release(sessionId, { ...params, status: "interrupted" });
|
|
}
|
|
|
|
function get(sessionId, params = {}) {
|
|
const session = sessions.get(requiredId(sessionId, "ses")) ?? null;
|
|
return session ? publicSession(session, params) : null;
|
|
}
|
|
|
|
function getByConversation(conversationId, params = {}) {
|
|
const sessionId = conversations.get(requiredId(conversationId, "cnv"));
|
|
return sessionId ? get(sessionId, { ...params, conversationId }) : null;
|
|
}
|
|
|
|
function expireIdle(params = {}) {
|
|
const timestamp = timestampFor(params.now ?? defaultNow);
|
|
const timestampMs = Date.parse(timestamp);
|
|
let expired = 0;
|
|
for (const session of sessions.values()) {
|
|
if (!sessionExpired(session, timestampMs)) continue;
|
|
session.status = "expired";
|
|
session.updatedAt = timestamp;
|
|
session.currentTraceId = null;
|
|
expired += 1;
|
|
}
|
|
return expired;
|
|
}
|
|
|
|
function describe() {
|
|
return {
|
|
kind: "code-agent-session-registry",
|
|
implementationType: "controlled-readonly-session-registry",
|
|
status: "available",
|
|
sessionCount: sessions.size,
|
|
maxSessions,
|
|
idleTimeoutMs,
|
|
statuses: [...CODE_AGENT_SESSION_STATUSES],
|
|
sessionMode: "controlled-readonly-session-registry",
|
|
recentSessions: [...sessions.values()]
|
|
.sort((a, b) => String(b.updatedAt).localeCompare(String(a.updatedAt)))
|
|
.slice(0, 10)
|
|
.map((session) => publicSession(session)),
|
|
longLivedCodexStdio: false,
|
|
longLivedSession: true,
|
|
durable: false,
|
|
secretMaterialStored: false,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function clear() {
|
|
sessions.clear();
|
|
conversations.clear();
|
|
}
|
|
|
|
function prune() {
|
|
if (sessions.size <= maxSessions) return;
|
|
const sorted = [...sessions.entries()]
|
|
.sort((a, b) => String(a[1].updatedAt).localeCompare(String(b[1].updatedAt)));
|
|
for (const [sessionId, session] of sorted.slice(0, sessions.size - maxSessions)) {
|
|
sessions.delete(sessionId);
|
|
for (const conversationId of session.conversationIds) {
|
|
if (conversations.get(conversationId) === sessionId) {
|
|
conversations.delete(conversationId);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return {
|
|
acquire,
|
|
release,
|
|
fail,
|
|
interrupt,
|
|
get,
|
|
getByConversation,
|
|
expireIdle,
|
|
describe,
|
|
clear
|
|
};
|
|
}
|
|
|
|
function blockedAcquire({ code, summary, session, timestamp, traceId }) {
|
|
const publicEvidence = session
|
|
? publicSession(session, { timestamp, reused: true })
|
|
: {
|
|
status: "failed",
|
|
updatedAt: timestamp,
|
|
lastTraceId: optionalId(traceId),
|
|
secretMaterialStored: false,
|
|
valuesRedacted: true
|
|
};
|
|
return {
|
|
ok: false,
|
|
code,
|
|
message: summary,
|
|
session: publicEvidence,
|
|
blocker: {
|
|
code,
|
|
sourceIssue: "pikasTech/HWLAB#317",
|
|
summary
|
|
}
|
|
};
|
|
}
|
|
|
|
function publicSession(session, { conversationId = null, reused = null } = {}) {
|
|
const conversationIds = [...session.conversationIds];
|
|
return {
|
|
sessionId: session.sessionId,
|
|
conversationId: conversationId ?? conversationIds[0] ?? null,
|
|
conversationIds,
|
|
status: session.status,
|
|
workspace: session.workspace,
|
|
sandbox: session.sandbox,
|
|
runnerKind: session.runnerKind,
|
|
sessionMode: session.sessionMode,
|
|
capabilityLevel: session.capabilityLevel,
|
|
implementationType: session.implementationType,
|
|
createdAt: session.createdAt,
|
|
updatedAt: session.updatedAt,
|
|
idleTimeoutMs: session.idleTimeoutMs,
|
|
expiresAt: session.expiresAt,
|
|
lastTraceId: session.lastTraceId,
|
|
currentTraceId: session.currentTraceId,
|
|
turn: session.turn,
|
|
reused: reused === null ? undefined : Boolean(reused),
|
|
durable: session.durable === true,
|
|
durableSession: session.durable === true,
|
|
longLivedSession: session.longLivedSession === true,
|
|
codexStdio: session.codexStdio === true,
|
|
readOnly: session.readOnly === true,
|
|
writeCapable: session.writeCapable === true,
|
|
secretMaterialStored: false,
|
|
valuesRedacted: true,
|
|
...(session.statusReason ? { statusReason: session.statusReason } : {})
|
|
};
|
|
}
|
|
|
|
function sessionExpired(session, timestampMs) {
|
|
if (!Number.isFinite(timestampMs) || session.status === "expired") return session.status === "expired";
|
|
return Date.parse(session.expiresAt) <= timestampMs;
|
|
}
|
|
|
|
function timestampFor(now) {
|
|
const value = typeof now === "function" ? now() : now;
|
|
if (typeof value === "string" && !Number.isNaN(Date.parse(value))) return value;
|
|
if (value instanceof Date && !Number.isNaN(value.valueOf())) return value.toISOString();
|
|
return new Date().toISOString();
|
|
}
|
|
|
|
function plusMs(timestampMs, offsetMs) {
|
|
return new Date(timestampMs + offsetMs).toISOString();
|
|
}
|
|
|
|
function requiredId(value, fallbackPrefix) {
|
|
const id = optionalId(value);
|
|
if (id) return id;
|
|
return `${fallbackPrefix}_${randomUUID()}`;
|
|
}
|
|
|
|
function optionalId(value) {
|
|
return typeof value === "string" && value.trim() ? value.trim() : null;
|
|
}
|
|
|
|
function positiveInteger(value, fallback) {
|
|
return Number.isInteger(value) && value > 0 ? value : fallback;
|
|
}
|