Compare commits
14 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e310ba4156 | |||
| ae7adb0847 | |||
| d24510172e | |||
| 29a4e709d0 | |||
| 8a08e6a6ee | |||
| c0dba8e904 | |||
| b91859c258 | |||
| f1433b05a6 | |||
| f64694f894 | |||
| e86b14a677 | |||
| 98f3821f91 | |||
| 21a02da00f | |||
| 346f5cc1df | |||
| ef73586a41 |
Generated
+2
-2
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "paperclip-adapter-claude-k8s",
|
||||
"version": "0.1.33",
|
||||
"version": "0.1.35",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "paperclip-adapter-claude-k8s",
|
||||
"version": "0.1.33",
|
||||
"version": "0.1.35",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"@kubernetes/client-node": "^1.0.0",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "paperclip-adapter-claude-k8s",
|
||||
"version": "0.1.33",
|
||||
"version": "0.1.35",
|
||||
"description": "Paperclip adapter plugin that runs Claude Code agents as Kubernetes Jobs",
|
||||
"license": "MIT",
|
||||
"repository": {
|
||||
|
||||
+5
-1
@@ -15,7 +15,6 @@ Core fields:
|
||||
- model (string, optional): Claude model id
|
||||
- effort (string, optional): reasoning effort passed via --effort (low|medium|high)
|
||||
- maxTurnsPerRun (number, optional): max turns for one run
|
||||
- dangerouslySkipPermissions (boolean, optional): pass --dangerously-skip-permissions to claude
|
||||
- instructionsFilePath (string, optional): absolute path to a markdown instructions file injected at runtime via --append-system-prompt-file
|
||||
- extraArgs (string[], optional): additional CLI args appended to the claude command
|
||||
- env (object, optional): KEY=VALUE environment variables; overrides inherited vars from the Deployment
|
||||
@@ -31,6 +30,11 @@ Kubernetes fields:
|
||||
- labels (object, optional): extra labels added to Job metadata
|
||||
- ttlSecondsAfterFinished (number, optional): auto-cleanup delay; default 300
|
||||
- retainJobs (boolean, optional): skip cleanup on completion for debugging
|
||||
- reattachOrphanedJobs (boolean, optional): when true (default), attach to a running orphaned Job that matches the current agent/task/session instead of blocking; when false, any non-terminal orphan blocks the new run
|
||||
|
||||
Output filtering fields:
|
||||
- enableRtk (boolean, optional): truncate oversized tool outputs before they reach the model via a PostToolUse hook; default false
|
||||
- rtkMaxOutputBytes (number, optional): byte threshold for tool output truncation when enableRtk is true; default 50000
|
||||
|
||||
Operational fields:
|
||||
- timeoutSec (number, optional): run timeout in seconds; 0 means no timeout
|
||||
|
||||
@@ -34,12 +34,10 @@ describe("getConfigSchema", () => {
|
||||
expect(field!.default).toBe(1000);
|
||||
});
|
||||
|
||||
it("dangerouslySkipPermissions defaults to true", () => {
|
||||
it("does not expose dangerouslySkipPermissions in UI schema", () => {
|
||||
const schema = getConfigSchema();
|
||||
const field = schema.fields.find((f: ConfigFieldSchema) => f.key === "dangerouslySkipPermissions");
|
||||
expect(field).toBeDefined();
|
||||
expect(field!.type).toBe("toggle");
|
||||
expect(field!.default).toBe(true);
|
||||
expect(field).toBeUndefined();
|
||||
});
|
||||
|
||||
it("reattachOrphanedJobs defaults to true", () => {
|
||||
|
||||
@@ -34,13 +34,6 @@ export function getConfigSchema(): AdapterConfigSchema {
|
||||
hint: "Maximum number of agentic turns (tool calls) per heartbeat run. 0 means unlimited.",
|
||||
default: 1000,
|
||||
},
|
||||
{
|
||||
type: "toggle",
|
||||
key: "dangerouslySkipPermissions",
|
||||
label: "Skip Permissions",
|
||||
hint: "Pass --dangerously-skip-permissions to Claude. Enabled by default for unattended K8s Jobs.",
|
||||
default: true,
|
||||
},
|
||||
// Kubernetes
|
||||
{
|
||||
type: "text",
|
||||
@@ -93,7 +86,7 @@ export function getConfigSchema(): AdapterConfigSchema {
|
||||
type: "toggle",
|
||||
key: "reattachOrphanedJobs",
|
||||
label: "Reattach to Orphaned Jobs",
|
||||
hint: "If a prior K8s Job for the same agent/task/session is still running (e.g. Paperclip restarted mid-run), attach to it and stream its output instead of deleting it and starting a new pod. Default: on.",
|
||||
hint: "If a prior K8s Job for the same agent/task/session is still running (e.g. Paperclip restarted mid-run), attach to it and stream its output instead of blocking the new run. When false, any non-terminal orphan blocks the new run. Default: on.",
|
||||
default: true,
|
||||
},
|
||||
// Resource Limits
|
||||
|
||||
+54
-35
@@ -15,7 +15,7 @@ vi.mock("./k8s-client.js", () => ({
|
||||
resetCache: vi.fn(),
|
||||
}));
|
||||
|
||||
const { isK8s404, buildPartialRunError, isReattachableOrphan, describePodTerminatedError, streamPodLogsOnce } = await import("./execute.js");
|
||||
const { isK8s404, buildPartialRunError, classifyOrphan, describePodTerminatedError, streamPodLogsOnce, execute } = await import("./execute.js");
|
||||
|
||||
function makeJob(opts: {
|
||||
runId?: string;
|
||||
@@ -146,59 +146,59 @@ describe("buildPartialRunError", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("isReattachableOrphan", () => {
|
||||
const agentId = "agent-abc";
|
||||
describe("classifyOrphan", () => {
|
||||
const taskId = "task-xyz";
|
||||
const sessionId = "sess-123";
|
||||
|
||||
it("returns true when agent/task/session all match and Job is not terminal", () => {
|
||||
const job = makeJob({ agentId, taskId, sessionId, runId: "old-run" });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(true);
|
||||
// --- Happy path: reattach ---
|
||||
it("returns reattach when taskId matches and both sessionIds match", () => {
|
||||
const job = makeJob({ taskId, sessionId });
|
||||
expect(classifyOrphan(job, { taskId, sessionId })).toBe("reattach");
|
||||
});
|
||||
|
||||
it("returns false when the Job is already Complete", () => {
|
||||
const job = makeJob({ agentId, taskId, sessionId, runId: "old-run", terminal: true });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(false);
|
||||
it("returns reattach when taskId matches and expected sessionId is null (missing on current side)", () => {
|
||||
const job = makeJob({ taskId, sessionId });
|
||||
expect(classifyOrphan(job, { taskId, sessionId: null })).toBe("reattach");
|
||||
});
|
||||
|
||||
it("returns false when expected taskId is null (caller couldn't derive one)", () => {
|
||||
const job = makeJob({ agentId, taskId, sessionId });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId: null, sessionId })).toBe(false);
|
||||
it("returns reattach when taskId matches and job has no session-id label (missing on job side)", () => {
|
||||
const job = makeJob({ taskId });
|
||||
expect(classifyOrphan(job, { taskId, sessionId })).toBe("reattach");
|
||||
});
|
||||
|
||||
it("returns false when expected sessionId is null", () => {
|
||||
const job = makeJob({ agentId, taskId, sessionId });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId: null })).toBe(false);
|
||||
it("returns reattach when taskId matches and neither side has a sessionId", () => {
|
||||
const job = makeJob({ taskId });
|
||||
expect(classifyOrphan(job, { taskId, sessionId: null })).toBe("reattach");
|
||||
});
|
||||
|
||||
it("returns false when agent id doesn't match", () => {
|
||||
const job = makeJob({ agentId: "agent-other", taskId, sessionId });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(false);
|
||||
// --- Block: task unknown ---
|
||||
it("returns block_task_unknown when expected taskId is null", () => {
|
||||
const job = makeJob({ taskId, sessionId });
|
||||
expect(classifyOrphan(job, { taskId: null, sessionId })).toBe("block_task_unknown");
|
||||
});
|
||||
|
||||
it("returns false when task id doesn't match", () => {
|
||||
const job = makeJob({ agentId, taskId: "task-other", sessionId });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(false);
|
||||
it("returns block_task_unknown when job has no task-id label", () => {
|
||||
const job = makeJob({ sessionId });
|
||||
expect(classifyOrphan(job, { taskId, sessionId })).toBe("block_task_unknown");
|
||||
});
|
||||
|
||||
it("returns false when session id doesn't match", () => {
|
||||
const job = makeJob({ agentId, taskId, sessionId: "sess-other" });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(false);
|
||||
// --- Block: task mismatch ---
|
||||
it("returns block_task_mismatch when both sides have taskId but they differ", () => {
|
||||
const job = makeJob({ taskId: "task-other", sessionId });
|
||||
expect(classifyOrphan(job, { taskId, sessionId })).toBe("block_task_mismatch");
|
||||
});
|
||||
|
||||
it("returns false when the Job is from a different adapter type", () => {
|
||||
const job = makeJob({ agentId, taskId, sessionId, adapterType: "claude_local" });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(false);
|
||||
// --- Block: session mismatch ---
|
||||
it("returns block_session_mismatch when taskId matches but sessionIds differ", () => {
|
||||
const job = makeJob({ taskId, sessionId: "sess-other" });
|
||||
expect(classifyOrphan(job, { taskId, sessionId })).toBe("block_session_mismatch");
|
||||
});
|
||||
|
||||
it("returns false when Job has no task-id label (labels were introduced in FAR-124)", () => {
|
||||
const job = makeJob({ agentId, sessionId });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false when Job has no session-id label", () => {
|
||||
const job = makeJob({ agentId, taskId });
|
||||
expect(isReattachableOrphan(job, { agentId, taskId, sessionId })).toBe(false);
|
||||
// --- Terminal orphan (caller filters these before classifyOrphan) ---
|
||||
it("returns reattach for terminal job (caller is responsible for filtering terminals)", () => {
|
||||
const job = makeJob({ taskId, sessionId, terminal: true });
|
||||
// classifyOrphan does not check terminal status — that is the caller's job
|
||||
expect(classifyOrphan(job, { taskId, sessionId })).toBe("reattach");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -261,6 +261,25 @@ describe("describePodTerminatedError", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("execute: all-invalid agent.id (N4)", () => {
|
||||
it("returns hard error without creating a Job when agent.id sanitizes to null", async () => {
|
||||
const logs: string[] = [];
|
||||
const result = await execute({
|
||||
runId: "run-001",
|
||||
agent: { id: "@@@", companyId: "co1", name: "Bad Agent", adapterType: "claude_k8s", adapterConfig: {} },
|
||||
runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
|
||||
config: {},
|
||||
context: {},
|
||||
onLog: async (_stream, msg) => { logs.push(msg); },
|
||||
});
|
||||
expect(result.errorCode).toBe("k8s_agent_id_invalid");
|
||||
expect(result.errorMessage).toContain("@@@");
|
||||
// getSelfPodInfo must NOT have been called (early return before K8s calls)
|
||||
const { getSelfPodInfo } = await import("./k8s-client.js");
|
||||
expect(getSelfPodInfo).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
// Regression: FAR-10 hardening — streamPodLogsOnce must not hang forever when
|
||||
// the K8s client's logApi.log call never resolves. When stopSignal fires, the
|
||||
// bail timer must force-return within LOG_STREAM_BAIL_TIMEOUT_MS (3s in the
|
||||
|
||||
+203
-66
@@ -89,30 +89,46 @@ export function buildPartialRunError(
|
||||
: `Claude exited with code ${exitCode ?? -1}`;
|
||||
}
|
||||
|
||||
export type OrphanClassification =
|
||||
| "reattach"
|
||||
| "block_session_mismatch"
|
||||
| "block_task_mismatch"
|
||||
| "block_task_unknown";
|
||||
|
||||
/**
|
||||
* Evaluate an orphaned K8s Job (one whose `paperclip.io/run-id` label does
|
||||
* not match the current runId) as a potential reattach target. A Job is
|
||||
* reattachable when it belongs to the same agent, same task, and same resume
|
||||
* session as the current run — meaning the previous Paperclip instance was
|
||||
* mid-stream on the exact piece of work this new run was dispatched to do.
|
||||
* Classify a non-terminal orphaned K8s Job (one whose `paperclip.io/run-id`
|
||||
* label does not match the current runId but does belong to this agent) as a
|
||||
* reattach candidate or a block reason.
|
||||
*
|
||||
* Decision matrix:
|
||||
* - taskId mismatch (both present, different values) → block_task_mismatch
|
||||
* - taskId missing on either side → block_task_unknown
|
||||
* - taskId match + both have sessionId + sessionIds differ → block_session_mismatch
|
||||
* - taskId match + one or both sides missing sessionId → reattach (reconcile)
|
||||
* - taskId match + both have sessionId + sessionIds match → reattach (happy path)
|
||||
*
|
||||
* Exported for unit tests.
|
||||
*/
|
||||
export function isReattachableOrphan(
|
||||
export function classifyOrphan(
|
||||
job: k8s.V1Job,
|
||||
expected: { agentId: string; taskId: string | null; sessionId: string | null },
|
||||
): boolean {
|
||||
if (!expected.taskId || !expected.sessionId) return false;
|
||||
expected: { taskId: string | null; sessionId: string | null },
|
||||
): OrphanClassification {
|
||||
const labels = job.metadata?.labels ?? {};
|
||||
if (labels["paperclip.io/adapter-type"] !== "claude_k8s") return false;
|
||||
if (labels["paperclip.io/agent-id"] !== expected.agentId) return false;
|
||||
if (labels["paperclip.io/task-id"] !== expected.taskId) return false;
|
||||
if (labels["paperclip.io/session-id"] !== expected.sessionId) return false;
|
||||
const conditions = job.status?.conditions ?? [];
|
||||
const terminal = conditions.some(
|
||||
(c) => (c.type === "Complete" || c.type === "Failed") && c.status === "True",
|
||||
);
|
||||
if (terminal) return false;
|
||||
return true;
|
||||
const jobTaskId = labels["paperclip.io/task-id"] ?? null;
|
||||
const jobSessionId = labels["paperclip.io/session-id"] ?? null;
|
||||
|
||||
// taskId missing on either side
|
||||
if (!expected.taskId || !jobTaskId) return "block_task_unknown";
|
||||
|
||||
// taskId mismatch
|
||||
if (expected.taskId !== jobTaskId) return "block_task_mismatch";
|
||||
|
||||
// taskId matches — check sessionId
|
||||
if (expected.sessionId && jobSessionId && expected.sessionId !== jobSessionId) {
|
||||
return "block_session_mismatch";
|
||||
}
|
||||
|
||||
return "reattach";
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -176,7 +192,7 @@ async function waitForPod(
|
||||
const containerStatuses = pod.status?.containerStatuses ?? [];
|
||||
|
||||
// Log phase transitions
|
||||
const statusKey = `${phase}:${initStatuses.map((s) => s.state?.waiting?.reason ?? s.state?.terminated?.reason ?? "ok").join(",")}:${containerStatuses.map((s) => s.state?.waiting?.reason ?? s.state?.running ? "running" : "waiting").join(",")}`;
|
||||
const statusKey = `${phase}:${initStatuses.map((s) => s.state?.waiting?.reason ?? s.state?.terminated?.reason ?? "ok").join(",")}:${containerStatuses.map((s) => s.state?.waiting?.reason ?? (s.state?.running ? "running" : "waiting")).join(",")}`;
|
||||
if (statusKey !== lastStatus) {
|
||||
const details: string[] = [`phase=${phase}`];
|
||||
for (const init of initStatuses) {
|
||||
@@ -301,7 +317,10 @@ export async function streamPodLogsOnce(
|
||||
if (stopSignal.stopped) {
|
||||
if (!writable.destroyed) writable.destroy();
|
||||
if (!bailTimer && bailResolve) {
|
||||
bailTimer = setTimeout(bailResolve, LOG_STREAM_BAIL_TIMEOUT_MS);
|
||||
bailTimer = setTimeout(() => {
|
||||
onLog("stderr", "[paperclip] Log stream bail timer fired — forcing return\n").catch(() => {});
|
||||
bailResolve!();
|
||||
}, LOG_STREAM_BAIL_TIMEOUT_MS);
|
||||
}
|
||||
}
|
||||
}, 200);
|
||||
@@ -345,6 +364,7 @@ async function streamPodLogs(
|
||||
onLog: AdapterExecutionContext["onLog"],
|
||||
kubeconfigPath?: string,
|
||||
stopSignal?: { stopped: boolean },
|
||||
dedup?: LogLineDedupFilter,
|
||||
): Promise<string> {
|
||||
const allChunks: string[] = [];
|
||||
let attempt = 0;
|
||||
@@ -354,7 +374,7 @@ async function streamPodLogs(
|
||||
let lastLogReceivedAt = Math.floor(Date.now() / 1000);
|
||||
// Shared across reconnects so replayed lines inside the `sinceSeconds`
|
||||
// overlap window are dropped before they reach the streaming UI (FAR-123).
|
||||
const dedup = new LogLineDedupFilter();
|
||||
if (!dedup) dedup = new LogLineDedupFilter();
|
||||
|
||||
while (!stopSignal?.stopped) {
|
||||
if (attempt >= MAX_LOG_RECONNECT_ATTEMPTS) {
|
||||
@@ -521,6 +541,17 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
// the current runId. When reattachOrphanedJobs is enabled and the orphan matches
|
||||
// the current agent+task+session, we attach to it instead of deleting it (FAR-124).
|
||||
const agentId = ctx.agent.id;
|
||||
const sanitizedAgentId = sanitizeLabelValue(agentId);
|
||||
if (!sanitizedAgentId) {
|
||||
await onLog("stderr", `[paperclip] Cannot create K8s Job: agent.id "${agentId}" produces no valid RFC 1123 label characters\n`);
|
||||
return {
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
errorMessage: `Agent ID "${agentId}" cannot be sanitized to a valid Kubernetes label`,
|
||||
errorCode: "k8s_agent_id_invalid",
|
||||
};
|
||||
}
|
||||
const selfPod = await getSelfPodInfo(kubeconfigPath);
|
||||
const guardNamespace = asString(config.namespace, "") || selfPod.namespace;
|
||||
const reattachOrphanedJobs = asBoolean(config.reattachOrphanedJobs, true);
|
||||
@@ -534,7 +565,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
const batchApi = getBatchApi(kubeconfigPath);
|
||||
const existing = await batchApi.listNamespacedJob({
|
||||
namespace: guardNamespace,
|
||||
labelSelector: `paperclip.io/agent-id=${agentId},paperclip.io/adapter-type=claude_k8s`,
|
||||
labelSelector: `paperclip.io/agent-id=${sanitizedAgentId},paperclip.io/adapter-type=claude_k8s`,
|
||||
});
|
||||
const running = existing.items.filter(
|
||||
(j) => !j.status?.conditions?.some((c) => (c.type === "Complete" || c.type === "Failed") && c.status === "True"),
|
||||
@@ -549,45 +580,72 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
(j) => (j.metadata?.labels?.["paperclip.io/run-id"] ?? "") === runId,
|
||||
);
|
||||
|
||||
// Pick the most recent reattachable orphan — same agent + task + session,
|
||||
// not terminal. Only one target is chosen; any other orphans get
|
||||
// cleaned up as before.
|
||||
if (reattachOrphanedJobs && orphaned.length > 0) {
|
||||
const candidates = orphaned
|
||||
.filter((j) =>
|
||||
isReattachableOrphan(j, {
|
||||
agentId,
|
||||
taskId: currentTaskLabel,
|
||||
sessionId: currentSessionLabel,
|
||||
}),
|
||||
)
|
||||
.sort((a, b) => {
|
||||
const at = new Date(a.metadata?.creationTimestamp ?? 0).getTime();
|
||||
const bt = new Date(b.metadata?.creationTimestamp ?? 0).getTime();
|
||||
return bt - at;
|
||||
});
|
||||
const chosen = candidates[0];
|
||||
const chosenName = chosen?.metadata?.name;
|
||||
if (chosen && chosenName) {
|
||||
reattachTarget = {
|
||||
jobName: chosenName,
|
||||
namespace: chosen.metadata?.namespace ?? guardNamespace,
|
||||
priorRunId: chosen.metadata?.labels?.["paperclip.io/run-id"] ?? "",
|
||||
image: chosen.spec?.template?.spec?.containers?.[0]?.image ?? "unknown",
|
||||
if (orphaned.length > 0) {
|
||||
if (!reattachOrphanedJobs) {
|
||||
// When reattach is disabled, block on any non-terminal orphan.
|
||||
const names = orphaned.map((j) => j.metadata?.name).join(", ");
|
||||
await onLog("stderr", `[paperclip] Concurrent run blocked: orphaned Job(s) running and reattach disabled: ${names}\n`);
|
||||
return {
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
errorMessage: `Concurrent run blocked: orphaned Job(s) still running for this agent (reattach disabled)`,
|
||||
errorCode: "k8s_concurrent_run_blocked",
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const toDelete = orphaned.filter(
|
||||
(j) => !reattachTarget || j.metadata?.name !== reattachTarget.jobName,
|
||||
);
|
||||
if (toDelete.length > 0) {
|
||||
const orphanNames = toDelete.map((j) => j.metadata?.name).join(", ");
|
||||
await onLog("stdout", `[paperclip] Cleaning up ${toDelete.length} orphaned K8s Job(s) from previous run(s): ${orphanNames}\n`);
|
||||
for (const j of toDelete) {
|
||||
const name = j.metadata?.name;
|
||||
if (name) {
|
||||
await cleanupJob(guardNamespace, name, onLog, kubeconfigPath);
|
||||
// Apply the decision matrix to each orphan, newest-first. The first
|
||||
// reattachable orphan becomes the target; any block classification
|
||||
// stops the new run immediately. Orphans are never deleted here —
|
||||
// terminal ones are cleaned up by TTL; live mismatches should not be
|
||||
// killed because they may still be doing real work.
|
||||
const sortedOrphans = [...orphaned].sort((a, b) => {
|
||||
const at = new Date(a.metadata?.creationTimestamp ?? 0).getTime();
|
||||
const bt = new Date(b.metadata?.creationTimestamp ?? 0).getTime();
|
||||
return bt - at;
|
||||
});
|
||||
for (const orphan of sortedOrphans) {
|
||||
const classification = classifyOrphan(orphan, {
|
||||
taskId: currentTaskLabel,
|
||||
sessionId: currentSessionLabel,
|
||||
});
|
||||
const orphanName = orphan.metadata?.name ?? "unknown";
|
||||
if (classification === "reattach") {
|
||||
if (!reattachTarget) {
|
||||
reattachTarget = {
|
||||
jobName: orphanName,
|
||||
namespace: orphan.metadata?.namespace ?? guardNamespace,
|
||||
priorRunId: orphan.metadata?.labels?.["paperclip.io/run-id"] ?? "",
|
||||
image: orphan.spec?.template?.spec?.containers?.[0]?.image ?? "unknown",
|
||||
};
|
||||
}
|
||||
} else if (classification === "block_task_unknown") {
|
||||
await onLog("stderr", `[paperclip] Blocked: orphaned Job ${orphanName} has missing task label — cannot safely reattach\n`);
|
||||
return {
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
errorMessage: `Concurrent run blocked: orphaned Job ${orphanName} has unknown task context`,
|
||||
errorCode: "k8s_orphan_task_unknown",
|
||||
};
|
||||
} else if (classification === "block_task_mismatch") {
|
||||
await onLog("stderr", `[paperclip] Blocked: orphaned Job ${orphanName} belongs to a different task\n`);
|
||||
return {
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
errorMessage: `Concurrent run blocked: orphaned Job ${orphanName} is running a different task`,
|
||||
errorCode: "k8s_concurrent_run_blocked",
|
||||
};
|
||||
} else if (classification === "block_session_mismatch") {
|
||||
await onLog("stderr", `[paperclip] Blocked: orphaned Job ${orphanName} has a different session\n`);
|
||||
return {
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
errorMessage: `Concurrent run blocked: orphaned Job ${orphanName} has a mismatched session`,
|
||||
errorCode: "k8s_orphan_session_mismatch",
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -686,6 +744,26 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
|
||||
await onLog("stdout", `[paperclip] Reattaching to in-flight K8s Job ${jobName} in namespace ${namespace} (prior run ${reattachTarget.priorRunId || "unknown"})\n`);
|
||||
|
||||
// Relabel the reattached Job with the current run-id (and session-id if
|
||||
// available) so the next concurrency guard sees it as owned by this run
|
||||
// rather than an orphan from the prior run.
|
||||
const labelPatch: Array<{ op: "add" | "replace"; path: string; value: string }> = [
|
||||
{ op: "replace", path: "/metadata/labels/paperclip.io~1run-id", value: runId },
|
||||
];
|
||||
if (currentSessionLabel) {
|
||||
labelPatch.push({ op: "replace", path: "/metadata/labels/paperclip.io~1session-id", value: currentSessionLabel });
|
||||
}
|
||||
try {
|
||||
await batchApi.patchNamespacedJob({
|
||||
name: jobName,
|
||||
namespace,
|
||||
body: labelPatch,
|
||||
});
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
await onLog("stderr", `[paperclip] Warning: failed to relabel reattached Job ${jobName}: ${msg}\n`);
|
||||
}
|
||||
} else {
|
||||
// Build Job manifest
|
||||
const built = buildJobManifest({ ctx, selfPod, promptBundle });
|
||||
@@ -696,6 +774,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
const claudeArgs = built.claudeArgs;
|
||||
const promptMetrics = built.promptMetrics;
|
||||
promptSecret = built.promptSecret;
|
||||
if (built.skippedLabels.length > 0) {
|
||||
await onLog("stderr", `[paperclip] Warning: skipped ${built.skippedLabels.length} extra label(s) with reserved prefix: ${built.skippedLabels.join(", ")}\n`);
|
||||
}
|
||||
|
||||
// Report invocation metadata
|
||||
if (onMeta) {
|
||||
@@ -751,11 +832,18 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
|
||||
// Create the Job
|
||||
let createdJobUid: string | undefined;
|
||||
try {
|
||||
await batchApi.createNamespacedJob({ namespace, body: job });
|
||||
const created = await batchApi.createNamespacedJob({ namespace, body: job });
|
||||
createdJobUid = created.metadata?.uid;
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
await onLog("stderr", `[paperclip] Failed to create K8s Job: ${msg}\n`);
|
||||
if (promptSecret) {
|
||||
try {
|
||||
await coreApi.deleteNamespacedSecret({ name: promptSecret.name, namespace: promptSecret.namespace });
|
||||
} catch { /* best-effort */ }
|
||||
}
|
||||
return {
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
@@ -765,6 +853,35 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
};
|
||||
}
|
||||
|
||||
// Attach ownerReference so K8s GC cleans up the Secret if the process
|
||||
// crashes before the finally block runs.
|
||||
if (promptSecret && createdJobUid) {
|
||||
try {
|
||||
await coreApi.patchNamespacedSecret({
|
||||
name: promptSecret.name,
|
||||
namespace: promptSecret.namespace,
|
||||
body: [
|
||||
{
|
||||
op: "add",
|
||||
path: "/metadata/ownerReferences",
|
||||
value: [
|
||||
{
|
||||
apiVersion: "batch/v1",
|
||||
kind: "Job",
|
||||
name: jobName,
|
||||
uid: createdJobUid,
|
||||
blockOwnerDeletion: false,
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
});
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
await onLog("stderr", `[paperclip] Warning: failed to set ownerReference on prompt Secret: ${msg}\n`);
|
||||
}
|
||||
}
|
||||
|
||||
await onLog("stdout", `[paperclip] Created K8s Job: ${jobName} in namespace ${namespace} (deadline: ${timeoutSec > 0 ? `${timeoutSec}s` : "none"})\n`);
|
||||
}
|
||||
|
||||
@@ -853,6 +970,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
let keepaliveTick = 0;
|
||||
let keepaliveJobTerminal = false;
|
||||
let keepaliveJobTerminalAt: number | null = null;
|
||||
let consecutiveTerminalReadings = 0;
|
||||
keepaliveTimer = setInterval(() => {
|
||||
// Fire-and-forget the async work; setInterval callbacks must be
|
||||
// synchronous or the timer will drift.
|
||||
@@ -875,19 +993,32 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
|
||||
// Verify the Job is still alive before announcing or refreshing.
|
||||
// Require two consecutive terminal readings before latching to
|
||||
// guard against a stale K8s API cache returning a false terminal
|
||||
// status on a single read (finding #5, FAR-15).
|
||||
try {
|
||||
const job = await batchApi.readNamespacedJob({ name: jobName, namespace });
|
||||
const terminal = job.status?.conditions?.some(
|
||||
(c) => (c.type === "Complete" || c.type === "Failed") && c.status === "True",
|
||||
);
|
||||
if (terminal) {
|
||||
keepaliveJobTerminal = true;
|
||||
keepaliveJobTerminalAt = Date.now();
|
||||
if (ctx.onSpawn) {
|
||||
consecutiveTerminalReadings++;
|
||||
if (consecutiveTerminalReadings >= 2) {
|
||||
keepaliveJobTerminal = true;
|
||||
keepaliveJobTerminalAt = Date.now();
|
||||
if (ctx.onSpawn) {
|
||||
void ctx.onSpawn({ pid: process.pid, processGroupId: null, startedAt: new Date().toISOString() }).catch(() => {});
|
||||
}
|
||||
return;
|
||||
}
|
||||
// First terminal reading — do not latch yet; next tick confirms.
|
||||
keepaliveTick++;
|
||||
if (ctx.onSpawn && (keepaliveTick === 1 || keepaliveTick % 12 === 0)) {
|
||||
void ctx.onSpawn({ pid: process.pid, processGroupId: null, startedAt: new Date().toISOString() }).catch(() => {});
|
||||
}
|
||||
return;
|
||||
}
|
||||
consecutiveTerminalReadings = 0;
|
||||
} catch (err: unknown) {
|
||||
// Only treat 404 (Job deleted) as terminal. Transient 5xx or
|
||||
// connection resets should NOT permanently disable the keepalive —
|
||||
@@ -928,9 +1059,12 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
// Shared signal: when job completion resolves, tell the log
|
||||
// streamer to stop reconnecting.
|
||||
const logStopSignal = { stopped: false };
|
||||
// Shared dedup filter: created here so the one-shot fallback can
|
||||
// reuse it and avoid pushing already-sent lines to the UI (finding #6, FAR-15).
|
||||
const logDedup = new LogLineDedupFilter();
|
||||
|
||||
const [logResult, completionResult] = await Promise.allSettled([
|
||||
streamPodLogs(namespace, podName, wrappedOnLog, kubeconfigPath, logStopSignal),
|
||||
streamPodLogs(namespace, podName, wrappedOnLog, kubeconfigPath, logStopSignal, logDedup),
|
||||
waitForJobCompletion(namespace, jobName, completionTimeoutMs, kubeconfigPath).then((r) => {
|
||||
logStopSignal.stopped = true;
|
||||
return r;
|
||||
@@ -958,7 +1092,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
// from the beginning of the log, giving us the full output.
|
||||
// We use a cheap string scan for the result-event guard (avoids a full JSON parse here;
|
||||
// the authoritative parse happens once below after all fallbacks complete).
|
||||
const hasResultEvent = stdout.includes('"type":"result"');
|
||||
const hasResultEvent = stdout.split("\n").some((l) => { try { return JSON.parse(l).type === "result"; } catch { return false; } });
|
||||
const needsOneShot = !stdout.trim() || (stdout.trim() && !hasResultEvent);
|
||||
if (needsOneShot) {
|
||||
if (!stdout.trim()) {
|
||||
@@ -967,9 +1101,12 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
const oneShotLogs = await readPodLogs(namespace, podName, kubeconfigPath);
|
||||
if (!stdout.trim() && oneShotLogs.trim()) {
|
||||
stdout = oneShotLogs;
|
||||
await onLog("stdout", stdout);
|
||||
const deduped = logDedup.filter(stdout) + logDedup.flush();
|
||||
if (deduped) await onLog("stdout", deduped);
|
||||
} else if (oneShotLogs && oneShotLogs.length > stdout.length) {
|
||||
await onLog("stdout", `[paperclip] Log stream captured partial output — supplemental one-shot read returned more content.\n`);
|
||||
const deduped = logDedup.filter(oneShotLogs) + logDedup.flush();
|
||||
if (deduped) await onLog("stdout", deduped);
|
||||
stdout = oneShotLogs;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -166,6 +166,75 @@ describe("buildJobManifest", () => {
|
||||
expect(job.metadata?.labels?.["paperclip.io/task-id"]).toBeUndefined();
|
||||
expect(job.metadata?.labels?.["paperclip.io/session-id"]).toBeUndefined();
|
||||
});
|
||||
|
||||
it("drops user label with paperclip.io/ prefix", () => {
|
||||
ctx.config = { labels: { "paperclip.io/run-id": "hijacked" } };
|
||||
const { job, skippedLabels } = buildJobManifest({ ctx, selfPod });
|
||||
expect(job.metadata?.labels?.["paperclip.io/run-id"]).not.toBe("hijacked");
|
||||
expect(skippedLabels).toContain("paperclip.io/run-id");
|
||||
});
|
||||
|
||||
it("drops user label with app.kubernetes.io/ prefix", () => {
|
||||
ctx.config = { labels: { "app.kubernetes.io/managed-by": "attacker" } };
|
||||
const { job, skippedLabels } = buildJobManifest({ ctx, selfPod });
|
||||
expect(job.metadata?.labels?.["app.kubernetes.io/managed-by"]).toBe("paperclip");
|
||||
expect(skippedLabels).toContain("app.kubernetes.io/managed-by");
|
||||
});
|
||||
|
||||
it("passes through user label without reserved prefix", () => {
|
||||
ctx.config = { labels: { "custom.io/team": "platform" } };
|
||||
const { job, skippedLabels } = buildJobManifest({ ctx, selfPod });
|
||||
expect(job.metadata?.labels?.["custom.io/team"]).toBe("platform");
|
||||
expect(skippedLabels).not.toContain("custom.io/team");
|
||||
});
|
||||
|
||||
it("populates skippedLabels with all dropped keys", () => {
|
||||
ctx.config = {
|
||||
labels: {
|
||||
"paperclip.io/agent-id": "x",
|
||||
"app.kubernetes.io/component": "y",
|
||||
"safe": "z",
|
||||
},
|
||||
};
|
||||
const { skippedLabels } = buildJobManifest({ ctx, selfPod });
|
||||
expect(skippedLabels).toHaveLength(2);
|
||||
expect(skippedLabels).toContain("paperclip.io/agent-id");
|
||||
expect(skippedLabels).toContain("app.kubernetes.io/component");
|
||||
});
|
||||
});
|
||||
|
||||
describe("system label sanitization (N4)", () => {
|
||||
it("sanitizes agent.id with @ to a valid RFC 1123 label", () => {
|
||||
ctx.agent.id = "user@example.com";
|
||||
const { job } = buildJobManifest({ ctx, selfPod });
|
||||
const label = job.metadata?.labels?.["paperclip.io/agent-id"];
|
||||
expect(label).toMatch(/^[a-zA-Z0-9]([a-zA-Z0-9._-]*[a-zA-Z0-9])?$/);
|
||||
expect(label).not.toContain("@");
|
||||
});
|
||||
|
||||
it("sanitizes agent.id with spaces to a valid RFC 1123 label", () => {
|
||||
ctx.agent.id = "my agent id";
|
||||
const { job } = buildJobManifest({ ctx, selfPod });
|
||||
const label = job.metadata?.labels?.["paperclip.io/agent-id"];
|
||||
expect(label).toMatch(/^[a-zA-Z0-9]([a-zA-Z0-9._-]*[a-zA-Z0-9])?$/);
|
||||
});
|
||||
|
||||
it("omits paperclip.io/run-id when sanitized value is null (all-invalid runId)", () => {
|
||||
// inject an all-special-chars runId via context override — buildJobManifest
|
||||
// uses ctx.runId directly
|
||||
const badCtx = makeCtx({ runId: "@@@" });
|
||||
const { job, skippedLabels } = buildJobManifest({ ctx: badCtx, selfPod });
|
||||
expect(job.metadata?.labels?.["paperclip.io/run-id"]).toBeUndefined();
|
||||
expect(skippedLabels).toContain("paperclip.io/run-id");
|
||||
});
|
||||
|
||||
it("selector matches sanitized agent-id label", () => {
|
||||
ctx.agent.id = "Agent@Test";
|
||||
const { job } = buildJobManifest({ ctx, selfPod });
|
||||
const agentLabel = job.metadata?.labels?.["paperclip.io/agent-id"];
|
||||
// the label should equal what sanitizeLabelValue produces
|
||||
expect(agentLabel).toBe("AgentTest");
|
||||
});
|
||||
});
|
||||
|
||||
describe("annotations", () => {
|
||||
@@ -438,10 +507,10 @@ describe("buildJobManifest", () => {
|
||||
|
||||
it("uses configured resource overrides", () => {
|
||||
ctx.config = {
|
||||
resources: {
|
||||
requests: { cpu: "500m", memory: "1Gi" },
|
||||
limits: { cpu: "2000m", memory: "4Gi" },
|
||||
},
|
||||
"resources.requests.cpu": "500m",
|
||||
"resources.requests.memory": "1Gi",
|
||||
"resources.limits.cpu": "2000m",
|
||||
"resources.limits.memory": "4Gi",
|
||||
};
|
||||
const { job } = buildJobManifest({ ctx, selfPod });
|
||||
const resources = job.spec?.template?.spec?.containers[0]?.resources;
|
||||
@@ -802,6 +871,28 @@ describe("buildJobManifest", () => {
|
||||
expect(filterScript).toContain("tool_result");
|
||||
});
|
||||
|
||||
it("filter script truncates without corrupting multi-byte UTF-8", () => {
|
||||
// "中" is U+4E2D, 3 bytes in UTF-8: E4 B8 AD
|
||||
// With MAX=5, two "中" (6 bytes) should truncate to one (3 bytes), not
|
||||
// produce a replacement character from slicing mid-codepoint.
|
||||
const setup = buildRtkSetupCommands(5);
|
||||
const b64Matches = [...setup.matchAll(/Buffer\.from\('([A-Za-z0-9+/=]+)','base64'\)/g)];
|
||||
const filterScript = Buffer.from(b64Matches[0]![1], "base64").toString("utf-8");
|
||||
|
||||
// Extract the trunc function from the filter script and evaluate it
|
||||
const fnMatch = filterScript.match(/(function trunc\(s\)\{.*\})(?=const tr=)/);
|
||||
expect(fnMatch).toBeTruthy();
|
||||
// eslint-disable-next-line no-eval
|
||||
const trunc = eval(`(()=>{const MAX=5;${fnMatch![1]};return trunc;})()`);
|
||||
|
||||
const result = trunc("中中");
|
||||
expect(result).not.toContain("�");
|
||||
expect(result).toContain("中");
|
||||
expect(result).toContain("truncated by paperclip-rtk");
|
||||
// Should report bytes from the actual truncation point, not MAX
|
||||
expect(result).toContain("3 bytes truncated");
|
||||
});
|
||||
|
||||
it("filter script handles array content (block format)", () => {
|
||||
const setup = buildRtkSetupCommands(50000);
|
||||
const b64Matches = [...setup.matchAll(/Buffer\.from\('([A-Za-z0-9+/=]+)','base64'\)/g)];
|
||||
|
||||
+30
-16
@@ -47,7 +47,8 @@ export function buildRtkSetupCommands(maxOutputBytes: number): string {
|
||||
`if(typeof s!=='string')return s;`,
|
||||
`const b=Buffer.from(s,'utf-8');`,
|
||||
`if(b.length<=MAX)return s;`,
|
||||
`return b.slice(0,MAX).toString('utf-8')+'\\n[...'+(b.length-MAX)+' bytes truncated by paperclip-rtk]';`,
|
||||
`let e=MAX;if(e>0){let p=e-1;while(p>0&&(b[p]&0xC0)===0x80)p--;const l=b[p];let n=1;if((l&0xE0)===0xC0)n=2;else if((l&0xF0)===0xE0)n=3;else if((l&0xF8)===0xF0)n=4;if(p+n>e)e=p;}`,
|
||||
`return b.slice(0,e).toString('utf-8')+'\\n[...'+(b.length-e)+' bytes truncated by paperclip-rtk]';`,
|
||||
`}`,
|
||||
`const tr=o&&(o.tool_response||o.tool_result);`,
|
||||
`if(tr){`,
|
||||
@@ -199,10 +200,14 @@ export interface JobBuildResult {
|
||||
/** Non-null when the prompt is too large for an env var and must be
|
||||
* staged as a K8s Secret before creating the Job. */
|
||||
promptSecret: PromptSecret | null;
|
||||
/** User-supplied extra labels that were dropped because they used a reserved prefix. */
|
||||
skippedLabels: string[];
|
||||
}
|
||||
|
||||
function sanitizeForK8sName(value: string, maxLen = 16): string {
|
||||
return value.toLowerCase().replace(/[^a-z0-9-]/g, "").slice(0, maxLen);
|
||||
// Trim trailing hyphens after slicing so names don't end with `-` when
|
||||
// truncation lands on a hyphen boundary (finding #16, FAR-15).
|
||||
return value.toLowerCase().replace(/[^a-z0-9-]/g, "").slice(0, maxLen).replace(/-+$/, "");
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -345,7 +350,6 @@ export function buildJobManifest(input: JobBuildInput): JobBuildResult {
|
||||
const extraArgs = asStringArray(config.extraArgs);
|
||||
const timeoutSec = asNumber(config.timeoutSec, 0);
|
||||
const ttlSeconds = asNumber(config.ttlSecondsAfterFinished, 300);
|
||||
const resources = parseObject(config.resources);
|
||||
const nodeSelector = parseKeyValueConfig(config.nodeSelector);
|
||||
const tolerations = Array.isArray(config.tolerations) ? config.tolerations : [];
|
||||
const extraLabels = parseKeyValueConfig(config.labels);
|
||||
@@ -427,29 +431,35 @@ export function buildJobManifest(input: JobBuildInput): JobBuildResult {
|
||||
// Build env vars
|
||||
const envVars = buildEnvVars(ctx, selfPod, config);
|
||||
|
||||
// Resource defaults
|
||||
const resourceRequests = parseObject(resources.requests);
|
||||
const resourceLimits = parseObject(resources.limits);
|
||||
// Resource defaults — UI stores dotted keys (e.g. "resources.requests.cpu")
|
||||
// as flat config entries, so read them directly from config with the dotted key.
|
||||
const containerResources: k8s.V1ResourceRequirements = {
|
||||
requests: {
|
||||
cpu: asString(resourceRequests.cpu, "1000m"),
|
||||
memory: asString(resourceRequests.memory, "2Gi"),
|
||||
cpu: asString(config["resources.requests.cpu"], "1000m"),
|
||||
memory: asString(config["resources.requests.memory"], "2Gi"),
|
||||
},
|
||||
limits: {
|
||||
cpu: asString(resourceLimits.cpu, "4000m"),
|
||||
memory: asString(resourceLimits.memory, "8Gi"),
|
||||
cpu: asString(config["resources.limits.cpu"], "4000m"),
|
||||
memory: asString(config["resources.limits.memory"], "8Gi"),
|
||||
},
|
||||
};
|
||||
|
||||
// Labels
|
||||
// Labels — system identifiers must pass RFC 1123 label value format.
|
||||
const sanitizedAgentId = sanitizeLabelValue(agent.id);
|
||||
const sanitizedRunId = sanitizeLabelValue(runId);
|
||||
const sanitizedCompanyId = sanitizeLabelValue(agent.companyId);
|
||||
const skippedLabels: string[] = [];
|
||||
if (!sanitizedRunId) skippedLabels.push("paperclip.io/run-id");
|
||||
if (!sanitizedCompanyId) skippedLabels.push("paperclip.io/company-id");
|
||||
const labels: Record<string, string> = {
|
||||
"app.kubernetes.io/managed-by": "paperclip",
|
||||
"app.kubernetes.io/component": "agent-job",
|
||||
"paperclip.io/agent-id": agent.id,
|
||||
"paperclip.io/run-id": runId,
|
||||
"paperclip.io/company-id": agent.companyId,
|
||||
// sanitizedAgentId null-check is enforced in execute.ts before Job creation
|
||||
"paperclip.io/agent-id": sanitizedAgentId ?? agent.id,
|
||||
"paperclip.io/adapter-type": "claude_k8s",
|
||||
};
|
||||
if (sanitizedRunId) labels["paperclip.io/run-id"] = sanitizedRunId;
|
||||
if (sanitizedCompanyId) labels["paperclip.io/company-id"] = sanitizedCompanyId;
|
||||
// Reattach-target labels: let a future execute() identify this Job as the
|
||||
// continuation of the same logical unit of work (same task + same resume
|
||||
// session) so it can attach to the running pod across a Paperclip restart
|
||||
@@ -460,7 +470,11 @@ export function buildJobManifest(input: JobBuildInput): JobBuildResult {
|
||||
const sessionLabel = runtimeSessionId ? sanitizeLabelValue(runtimeSessionId) : null;
|
||||
if (sessionLabel) labels["paperclip.io/session-id"] = sessionLabel;
|
||||
for (const [key, value] of Object.entries(extraLabels)) {
|
||||
labels[key] = value;
|
||||
if (key.startsWith("paperclip.io/") || key.startsWith("app.kubernetes.io/")) {
|
||||
skippedLabels.push(key);
|
||||
} else {
|
||||
labels[key] = value;
|
||||
}
|
||||
}
|
||||
|
||||
// Volumes
|
||||
@@ -627,5 +641,5 @@ export function buildJobManifest(input: JobBuildInput): JobBuildResult {
|
||||
},
|
||||
};
|
||||
|
||||
return { job, jobName, namespace, prompt, claudeArgs, promptMetrics, promptSecret };
|
||||
return { job, jobName, namespace, prompt, claudeArgs, promptMetrics, promptSecret, skippedLabels };
|
||||
}
|
||||
|
||||
@@ -106,7 +106,12 @@ export async function getSelfPodInfo(kubeconfigPath?: string): Promise<SelfPodIn
|
||||
throw new Error(`claude_k8s: pod ${hostname} has no spec`);
|
||||
}
|
||||
|
||||
const mainContainer = spec.containers[0];
|
||||
// Match the Paperclip container by name ("paperclip") to avoid service-mesh
|
||||
// sidecars or other injected containers being picked up as the source of
|
||||
// truth for the Job spec (finding #9, FAR-15). Fall back to the first
|
||||
// container if no name match is found (matches prior behavior).
|
||||
const mainContainer =
|
||||
spec.containers.find((c) => c.name === "paperclip") ?? spec.containers[0];
|
||||
if (!mainContainer?.image) {
|
||||
throw new Error(`claude_k8s: pod ${hostname} has no container image`);
|
||||
}
|
||||
|
||||
+15
-6
@@ -9,9 +9,12 @@ export function parseClaudeStreamJson(stdout: string) {
|
||||
let model = "";
|
||||
let finalResult: Record<string, unknown> | null = null;
|
||||
const assistantTexts: string[] = [];
|
||||
// Belt-and-braces dedup: track seen text blocks to filter duplicates
|
||||
// caused by log stream reconnects replaying overlapping windows.
|
||||
const seenTexts = new Set<string>();
|
||||
// Belt-and-braces dedup: key by (message.id, textIndex) so a session that
|
||||
// legitimately emits the same text twice in different turns isn't collapsed
|
||||
// (finding #11, FAR-15). The log-dedup filter handles reconnect overlaps
|
||||
// at the line level; this guard only needs to protect against the same
|
||||
// message block being parsed twice.
|
||||
const seenBlocks = new Set<string>();
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
@@ -29,14 +32,20 @@ export function parseClaudeStreamJson(stdout: string) {
|
||||
if (type === "assistant") {
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
const message = parseObject(event.message);
|
||||
const messageId = asString(message.id, "");
|
||||
const content = Array.isArray(message.content) ? message.content : [];
|
||||
for (const entry of content) {
|
||||
for (let i = 0; i < content.length; i++) {
|
||||
const entry = content[i];
|
||||
if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue;
|
||||
const block = entry as Record<string, unknown>;
|
||||
if (asString(block.type, "") === "text") {
|
||||
const text = asString(block.text, "");
|
||||
if (text && !seenTexts.has(text)) {
|
||||
seenTexts.add(text);
|
||||
if (!text) continue;
|
||||
// Prefer (messageId, index) when the message has an id; fall back
|
||||
// to text content when it doesn't (legacy/partial events).
|
||||
const key = messageId ? `${messageId}:${i}` : `text:${text}`;
|
||||
if (!seenBlocks.has(key)) {
|
||||
seenBlocks.add(key);
|
||||
assistantTexts.push(text);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { describe, it, expect, vi } from "vitest";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { prepareClaudePromptBundle } from "./prompt-cache.js";
|
||||
|
||||
const onLog = vi.fn();
|
||||
|
||||
describe("prepareClaudePromptBundle path traversal validation", () => {
|
||||
const validArgs = {
|
||||
skills: [],
|
||||
instructionsContents: null,
|
||||
onLog,
|
||||
};
|
||||
|
||||
it("rejects companyId containing ..", async () => {
|
||||
await expect(prepareClaudePromptBundle({ ...validArgs, companyId: ".." })).rejects.toThrow(/companyId/);
|
||||
});
|
||||
|
||||
it("rejects companyId containing ../x", async () => {
|
||||
await expect(prepareClaudePromptBundle({ ...validArgs, companyId: "../x" })).rejects.toThrow(/companyId/);
|
||||
});
|
||||
|
||||
it("rejects companyId containing /", async () => {
|
||||
await expect(prepareClaudePromptBundle({ ...validArgs, companyId: "a/b" })).rejects.toThrow(/companyId/);
|
||||
});
|
||||
|
||||
it("rejects companyId containing backslash", async () => {
|
||||
await expect(prepareClaudePromptBundle({ ...validArgs, companyId: "a\\b" })).rejects.toThrow(/companyId/);
|
||||
});
|
||||
|
||||
it("rejects companyId containing null byte", async () => {
|
||||
await expect(prepareClaudePromptBundle({ ...validArgs, companyId: "a\0b" })).rejects.toThrow(/companyId/);
|
||||
});
|
||||
|
||||
it("rejects empty companyId", async () => {
|
||||
await expect(prepareClaudePromptBundle({ ...validArgs, companyId: "" })).rejects.toThrow(/companyId/);
|
||||
});
|
||||
|
||||
it("rejects whitespace-only companyId", async () => {
|
||||
await expect(prepareClaudePromptBundle({ ...validArgs, companyId: " " })).rejects.toThrow(/companyId/);
|
||||
});
|
||||
|
||||
it("accepts a valid companyId", async () => {
|
||||
vi.stubEnv("PAPERCLIP_HOME", path.join(os.tmpdir(), `prompt-cache-test-${process.pid}`));
|
||||
const result = await prepareClaudePromptBundle({ ...validArgs, companyId: "acme-co" });
|
||||
expect(result.rootDir).toContain("acme-co");
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
});
|
||||
@@ -21,6 +21,13 @@ export interface ClaudePromptBundle {
|
||||
|
||||
const DEFAULT_PAPERCLIP_INSTANCE_ID = "default";
|
||||
|
||||
function validatePathComponent(value: string, fieldName: string): void {
|
||||
if (value.trim().length === 0) throw new Error(`Invalid ${fieldName}: must not be empty`);
|
||||
if (value.includes("/") || value.includes("\\")) throw new Error(`Invalid ${fieldName}: must not contain path separators`);
|
||||
if (value.includes("..")) throw new Error(`Invalid ${fieldName}: must not contain ".."`);
|
||||
if (value.includes("\0")) throw new Error(`Invalid ${fieldName}: must not contain null bytes`);
|
||||
}
|
||||
|
||||
function resolveManagedClaudePromptCacheRoot(companyId: string): string {
|
||||
const paperclipHome =
|
||||
(typeof process.env.PAPERCLIP_HOME === "string" && process.env.PAPERCLIP_HOME.trim().length > 0
|
||||
@@ -31,6 +38,8 @@ function resolveManagedClaudePromptCacheRoot(companyId: string): string {
|
||||
(typeof process.env.PAPERCLIP_INSTANCE_ID === "string" && process.env.PAPERCLIP_INSTANCE_ID.trim().length > 0
|
||||
? process.env.PAPERCLIP_INSTANCE_ID.trim()
|
||||
: null) ?? DEFAULT_PAPERCLIP_INSTANCE_ID;
|
||||
validatePathComponent(companyId, "companyId");
|
||||
validatePathComponent(instanceId, "instanceId");
|
||||
return path.resolve(paperclipHome, "instances", instanceId, "companies", companyId, "claude-prompt-cache");
|
||||
}
|
||||
|
||||
|
||||
+16
-9
@@ -85,8 +85,13 @@ async function checkRbac(
|
||||
{ resource: "jobs", group: "batch", verb: "create", code: "k8s_rbac_job_create", label: "create Jobs" },
|
||||
{ resource: "jobs", group: "batch", verb: "delete", code: "k8s_rbac_job_delete", label: "delete Jobs" },
|
||||
{ resource: "jobs", group: "batch", verb: "get", code: "k8s_rbac_job_get", label: "get Jobs" },
|
||||
{ resource: "jobs", group: "batch", verb: "list", code: "k8s_rbac_job_list", label: "list Jobs" },
|
||||
{ resource: "pods", group: "", verb: "list", code: "k8s_rbac_pod_list", label: "list Pods" },
|
||||
{ resource: "pods/log", group: "", verb: "get", code: "k8s_rbac_pod_log", label: "get Pod logs" },
|
||||
{ resource: "secrets", group: "", verb: "create", code: "k8s_rbac_secret_create", label: "create Secrets" },
|
||||
{ resource: "secrets", group: "", verb: "delete", code: "k8s_rbac_secret_delete", label: "delete Secrets" },
|
||||
{ resource: "secrets", group: "", verb: "get", code: "k8s_rbac_secret_get", label: "get Secrets" },
|
||||
{ resource: "persistentvolumeclaims", group: "", verb: "get", code: "k8s_rbac_pvc_get", label: "get PersistentVolumeClaims" },
|
||||
];
|
||||
|
||||
for (const check of rbacChecks) {
|
||||
@@ -221,16 +226,18 @@ export async function testEnvironment(
|
||||
|
||||
// 2. Target namespace exists
|
||||
const nsOk = await checkNamespace(namespace, selfPod.namespace, checks, kubeconfigPath);
|
||||
if (!nsOk) {
|
||||
return { adapterType: ctx.adapterType, status: summarizeStatus(checks), checks, testedAt: new Date().toISOString() };
|
||||
}
|
||||
|
||||
// 3-5. Run remaining checks in parallel
|
||||
await Promise.all([
|
||||
checkRbac(namespace, checks, kubeconfigPath),
|
||||
checkSecret(namespace, secretRef, checks, kubeconfigPath),
|
||||
checkPvc(selfPod, checks, kubeconfigPath),
|
||||
]);
|
||||
// 3-5. Run remaining checks even if namespace check failed so operators see
|
||||
// all issues at once instead of fixing them one at a time.
|
||||
if (nsOk) {
|
||||
await Promise.all([
|
||||
checkRbac(namespace, checks, kubeconfigPath),
|
||||
checkSecret(namespace, secretRef, checks, kubeconfigPath),
|
||||
checkPvc(selfPod, checks, kubeconfigPath),
|
||||
]);
|
||||
} else {
|
||||
await checkRbac(namespace, checks, kubeconfigPath);
|
||||
}
|
||||
|
||||
return {
|
||||
adapterType: ctx.adapterType,
|
||||
|
||||
Reference in New Issue
Block a user