Merge remote-tracking branch 'origin/main' into latest2-1714
# Conflicts: # packages/engine/src/agent-tools.ts # packages/engine/src/base-commit-capture.ts # packages/engine/src/worktree-acquisition.ts
This commit is contained in:
5
.changeset/session-routing-headers.md
Normal file
5
.changeset/session-routing-headers.md
Normal file
@@ -0,0 +1,5 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
Add `X-Session-Id` and `X-Session-Affinity` request headers to all LLM chat completion requests. These let LLM gateways sticky-route consecutive requests from the same conversation to the same backend, and let observability tools (Langfuse, Arize, etc.) group the otherwise-stateless API calls of a session into a single multi-turn trace. Both headers carry the same stable identifier — the task id when available (stable across pause/resume), otherwise the pi session id. (#1675)
|
||||
@@ -631,8 +631,12 @@ function areTaskCardPropsEqual(previous: TaskCardProps, next: TaskCardProps): bo
|
||||
// F7 — compare the sorted key SETS, not just the count: a same-count repo swap (one
|
||||
// repo released, a different one acquired) keeps the count but must still re-render,
|
||||
// otherwise the placeholder shows a stale repo set.
|
||||
JSON.stringify(Object.keys(previousTask.workspaceWorktrees ?? {}).sort()) ===
|
||||
JSON.stringify(Object.keys(nextTask.workspaceWorktrees ?? {}).sort()) &&
|
||||
// FNXC:Workspace 2026-06-22-09:00: compare full VALUES, not only the key set. A
|
||||
// pool-reclaim re-acquire keeps the same repo key but produces a different
|
||||
// worktreePath/branch; a key-set-only check would leave the card showing stale path
|
||||
// text. Whole-map JSON compare covers keys and values at negligible cost for small N.
|
||||
JSON.stringify(previousTask.workspaceWorktrees ?? null) ===
|
||||
JSON.stringify(nextTask.workspaceWorktrees ?? null) &&
|
||||
previousTask.branch === nextTask.branch &&
|
||||
previousTask.baseBranch === nextTask.baseBranch &&
|
||||
previousTask.breakIntoSubtasks === nextTask.breakIntoSubtasks &&
|
||||
|
||||
@@ -3148,7 +3148,11 @@ export function TaskDetailContent({
|
||||
{/* FNXC:Workspace 2026-06-21-00:00: workspace tasks have no singular
|
||||
task.worktree/task.branch; surface their acquired per-sub-repo worktrees
|
||||
as a flat read-only list so the detail view isn't blank (U3/KTD5). */}
|
||||
{isWorkspaceTask(task) && <WorkspaceWorktreesSummary task={task} />}
|
||||
{/* FNXC:Workspace 2026-06-22-09:00: gate/render off the hydrated
|
||||
workingTask, not the sparse task row. workspaceWorktrees is only
|
||||
present in fetched detail, so keying off task renders blank on the
|
||||
optimistic-open path before the detail fetch resolves. */}
|
||||
{isWorkspaceTask(workingTask) && <WorkspaceWorktreesSummary task={workingTask} />}
|
||||
</>
|
||||
)}
|
||||
{task.status === "failed" && task.error && (
|
||||
|
||||
@@ -48,8 +48,9 @@ describeIfGit("workspace fixture", () => {
|
||||
|
||||
it("builds a non-git root with two real git sub-repos and a resolvable workspace config", async () => {
|
||||
fx = await createWorkspaceFixture();
|
||||
// Root is NOT a git repo.
|
||||
expect(() => fx.git("..", "git rev-parse --git-dir")).toThrow();
|
||||
// Root is NOT a git repo. Use "." so the check runs in fx.rootDir itself, not
|
||||
// its parent (".." would resolve to the tmpdir and could pass spuriously).
|
||||
expect(() => fx.git(".", "git rev-parse --git-dir")).toThrow();
|
||||
// Each sub-repo is a real git repo with a commit on main.
|
||||
expect(fx.git("repo-a", "git rev-parse --abbrev-ref HEAD")).toBe("main");
|
||||
expect(fx.git("repo-b", "git rev-list --count HEAD")).toBe("1");
|
||||
|
||||
@@ -15,6 +15,11 @@ const findMock = vi.fn();
|
||||
const getAllMock = vi.fn(() => [] as any[]);
|
||||
const registerProviderMock = vi.fn();
|
||||
const refreshMock = vi.fn();
|
||||
// FNXC:SessionRouting 2026-06-24-11:30:
|
||||
// #1675: capture model-registry auth resolution + session id so the wiring
|
||||
// test can assert X-Session-Id/X-Session-Affinity precedence end-to-end.
|
||||
const getApiKeyAndHeadersMock = vi.fn(async () => ({ ok: true, apiKey: undefined, headers: undefined }));
|
||||
const sessionManagerGetSessionIdMock = vi.fn(() => undefined);
|
||||
const settingsManagerCreateMock = vi.fn(() => ({ kind: "settings-manager-create" }));
|
||||
const settingsManagerInMemoryMock = vi.fn(() => ({ kind: "settings-manager" }));
|
||||
const setFallbackResolverMock = vi.fn();
|
||||
@@ -138,9 +143,12 @@ vi.mock("@earendil-works/pi-coding-agent", () => ({
|
||||
refresh() {
|
||||
return refreshMock();
|
||||
}
|
||||
getApiKeyAndHeaders() {
|
||||
return getApiKeyAndHeadersMock();
|
||||
}
|
||||
},
|
||||
SessionManager: {
|
||||
inMemory: () => ({ kind: "session-manager" }),
|
||||
inMemory: () => ({ kind: "session-manager", getSessionId: sessionManagerGetSessionIdMock }),
|
||||
},
|
||||
SettingsManager: {
|
||||
create: settingsManagerCreateMock,
|
||||
@@ -1024,6 +1032,9 @@ describe("createFnAgent", () => {
|
||||
realpathSyncNativeMock.mockImplementation((path: PathLike) => String(path));
|
||||
readCustomProvidersMock.mockReturnValue([]);
|
||||
findMock.mockImplementation((provider: string, modelId: string) => ({ provider, id: modelId }));
|
||||
// #1675: re-establish default auth + session-id mock returns after clearAllMocks.
|
||||
getApiKeyAndHeadersMock.mockResolvedValue({ ok: true, apiKey: undefined, headers: undefined });
|
||||
sessionManagerGetSessionIdMock.mockReturnValue(undefined);
|
||||
createBashToolMock.mockClear();
|
||||
createAgentSessionMock.mockResolvedValue({
|
||||
session: {
|
||||
@@ -1921,6 +1932,62 @@ describe("createFnAgent", () => {
|
||||
warnSpy.mockRestore();
|
||||
});
|
||||
|
||||
// FNXC:SessionRouting 2026-06-24-11:30:
|
||||
// #1675: createFnAgent must resolve sessionRoutingId = taskId ?? piSessionId and
|
||||
// wrap the registry's getApiKeyAndHeaders so outbound requests carry routing
|
||||
// headers. These assert the wiring precedence end-to-end, not just the helper.
|
||||
describe("session routing headers wiring (#1675)", () => {
|
||||
const anyModel = { provider: "anthropic", id: "claude" } as never;
|
||||
|
||||
async function createAndCaptureRegistry(overrides: Record<string, unknown> = {}) {
|
||||
const { createFnAgent } = await import("../pi.js");
|
||||
await createFnAgent({
|
||||
cwd: "/tmp",
|
||||
systemPrompt: "test",
|
||||
tools: "readonly",
|
||||
...overrides,
|
||||
});
|
||||
const sessionOptions = createAgentSessionMock.mock.calls.at(-1)?.[0] as {
|
||||
modelRegistry: { getApiKeyAndHeaders: (model: unknown) => Promise<unknown> };
|
||||
};
|
||||
return sessionOptions.modelRegistry;
|
||||
}
|
||||
|
||||
it("uses taskId as the routing id when provided", async () => {
|
||||
const registry = await createAndCaptureRegistry({ taskId: "FN-7788" });
|
||||
|
||||
const result = await registry.getApiKeyAndHeaders(anyModel) as { ok: boolean; headers?: Record<string, string> };
|
||||
|
||||
expect(result.ok).toBe(true);
|
||||
expect(result.headers).toEqual({
|
||||
"X-Session-Id": "FN-7788",
|
||||
"X-Session-Affinity": "FN-7788",
|
||||
});
|
||||
});
|
||||
|
||||
it("falls back to the pi session id when taskId is absent", async () => {
|
||||
sessionManagerGetSessionIdMock.mockReturnValue("pi-session-abc");
|
||||
const registry = await createAndCaptureRegistry();
|
||||
|
||||
const result = await registry.getApiKeyAndHeaders(anyModel) as { ok: boolean; headers?: Record<string, string> };
|
||||
|
||||
expect(result.headers).toEqual({
|
||||
"X-Session-Id": "pi-session-abc",
|
||||
"X-Session-Affinity": "pi-session-abc",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not wrap getApiKeyAndHeaders when neither taskId nor a session id is available", async () => {
|
||||
// getApiKeyAndHeadersMock returns { ok: true, headers: undefined }; if the
|
||||
// wrapper were applied, headers would be populated with X-Session-*.
|
||||
const registry = await createAndCaptureRegistry();
|
||||
|
||||
const result = await registry.getApiKeyAndHeaders(anyModel) as { ok: boolean; headers?: Record<string, string> };
|
||||
|
||||
expect(result.headers).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("skill selection", () => {
|
||||
beforeEach(() => {
|
||||
// Reset modules to ensure fresh imports for each test
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import type { ModelRegistry } from "@earendil-works/pi-coding-agent";
|
||||
import { attachSessionRoutingHeaders, buildSessionRoutingHeaders } from "../pi.js";
|
||||
|
||||
// FNXC:SessionRouting 2026-06-23-16:40:
|
||||
// Issue #1675: chat completion requests must carry X-Session-Id and
|
||||
// X-Session-Affinity so LLM gateways can sticky-route and observability tools
|
||||
// can group the stateless API calls of one conversation into a single trace.
|
||||
|
||||
describe("buildSessionRoutingHeaders", () => {
|
||||
it("emits X-Session-Id and X-Session-Affinity with the same identifier", () => {
|
||||
expect(buildSessionRoutingHeaders("sess-123")).toEqual({
|
||||
"X-Session-Id": "sess-123",
|
||||
"X-Session-Affinity": "sess-123",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("attachSessionRoutingHeaders", () => {
|
||||
// Minimal stand-in for the bits of ModelRegistry the wrapper touches.
|
||||
function makeRegistry(
|
||||
resolve: (model: unknown) => Promise<{ ok: boolean; apiKey?: string; headers?: Record<string, string>; error?: string }>,
|
||||
): ModelRegistry {
|
||||
return { getApiKeyAndHeaders: resolve } as unknown as ModelRegistry;
|
||||
}
|
||||
|
||||
const anyModel = { provider: "anthropic", id: "claude" } as never;
|
||||
|
||||
it("merges the routing headers into resolved request headers", async () => {
|
||||
const registry = makeRegistry(async () => ({ ok: true, apiKey: "sk-live", headers: undefined }));
|
||||
attachSessionRoutingHeaders(registry, "sess-abc");
|
||||
|
||||
const result = await registry.getApiKeyAndHeaders(anyModel);
|
||||
|
||||
expect(result).toEqual({
|
||||
ok: true,
|
||||
apiKey: "sk-live",
|
||||
headers: {
|
||||
"X-Session-Id": "sess-abc",
|
||||
"X-Session-Affinity": "sess-abc",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves the resolved apiKey and any provider-specific headers", async () => {
|
||||
const registry = makeRegistry(async () => ({
|
||||
ok: true,
|
||||
apiKey: "sk-custom",
|
||||
headers: { "HTTP-Referer": "https://example.com", "X-Title": "Fusion" },
|
||||
}));
|
||||
attachSessionRoutingHeaders(registry, "sess-xyz");
|
||||
|
||||
const result = await registry.getApiKeyAndHeaders(anyModel);
|
||||
|
||||
expect(result.ok).toBe(true);
|
||||
if (!result.ok) throw new Error("expected ok auth result");
|
||||
expect(result.apiKey).toBe("sk-custom");
|
||||
expect(result.headers).toEqual({
|
||||
"HTTP-Referer": "https://example.com",
|
||||
"X-Title": "Fusion",
|
||||
"X-Session-Id": "sess-xyz",
|
||||
"X-Session-Affinity": "sess-xyz",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not alter failed auth resolutions", async () => {
|
||||
const registry = makeRegistry(async () => ({ ok: false, error: "No API key found" }));
|
||||
attachSessionRoutingHeaders(registry, "sess-fail");
|
||||
|
||||
const result = await registry.getApiKeyAndHeaders(anyModel);
|
||||
|
||||
expect(result).toEqual({ ok: false, error: "No API key found" });
|
||||
});
|
||||
|
||||
it("no-ops without throwing when getApiKeyAndHeaders is absent", () => {
|
||||
// If a future pi-coding-agent rename removes the method, the wrapper must not
|
||||
// break session creation. It leaves the registry untouched and warns instead.
|
||||
const registry = {} as ModelRegistry;
|
||||
|
||||
expect(() => attachSessionRoutingHeaders(registry, "sess-none")).not.toThrow();
|
||||
expect((registry as unknown as Record<string, unknown>).getApiKeyAndHeaders).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -3921,6 +3921,15 @@ export function createAcquireRepoWorktreeTool(opts: {
|
||||
};
|
||||
}
|
||||
const freshTask = await store.getTask(task.id);
|
||||
/*
|
||||
FNXC:Workspace 2026-06-21-22:30:
|
||||
F1 — acquireWorkspaceRepoWorktree can throw WorkspaceRepoAcquireBusyError on
|
||||
same-sub-repo contention (KTD4) or a generic failure. Both must surface as a
|
||||
structured isError tool result, never an uncaught throw that crashes the agent
|
||||
loop. The busy message is sanitized — it does NOT leak the holder task id into
|
||||
agent-facing text (only into details). runContext is forwarded so the helper's
|
||||
audit/log entries keep run attribution.
|
||||
*/
|
||||
let result: Awaited<ReturnType<typeof acquireWorkspaceRepoWorktree>>;
|
||||
try {
|
||||
result = await acquireWorkspaceRepoWorktree({
|
||||
@@ -3951,10 +3960,14 @@ export function createAcquireRepoWorktreeTool(opts: {
|
||||
isError: true,
|
||||
};
|
||||
}
|
||||
// FNXC:Workspace 2026-06-21-22:30: F2 — register a freshly-acquired sub-repo worktree in the executor's activeWorktrees Set (skip the already-acquired short-circuit; that path was registered on its original fresh acquire).
|
||||
if (!result.alreadyAcquired) {
|
||||
onAcquired?.(result.worktreePath);
|
||||
}
|
||||
// FNXC:Workspace 2026-06-21-22:30: F2 — register a freshly-acquired sub-repo worktree in the executor's activeWorktrees Set (KTD2) so owner/liveness checks see live per-repo worktrees, not just the browse-only root.
|
||||
// FNXC:Workspace 2026-06-22-09:00: register UNCONDITIONALLY, including the
|
||||
// already-acquired short-circuit. After an executor restart activeWorktrees is an
|
||||
// empty Map; a resumed workspace task with pre-existing task.workspaceWorktrees hits
|
||||
// the alreadyAcquired path, so skipping onAcquired left the sub-repo path unregistered
|
||||
// in-memory and conflict/liveness checks missed it. Set.add is idempotent, so re-firing
|
||||
// on a fresh acquire is a harmless no-op.
|
||||
onAcquired?.(result.worktreePath);
|
||||
await store.logEntry(
|
||||
task.id,
|
||||
result.alreadyAcquired
|
||||
|
||||
@@ -39,16 +39,19 @@ export async function resolveCapturedBaseCommitSha(
|
||||
integrationBranch: string = "main",
|
||||
): Promise<string | undefined> {
|
||||
const branch = integrationBranch.trim() || "main";
|
||||
// FNXC:Workspace 2026-06-22-00:00:
|
||||
// Shell-quote with POSIX single quotes, NOT JSON.stringify. JSON.stringify wraps
|
||||
// in double quotes, under which the shell expands `$VAR`/backticks — a branch like
|
||||
// `release/$2.0` would expand `$2` to a positional. Admin-configured integration
|
||||
// branch names are not guaranteed to exclude `$`, and `$` is valid in git refs, so
|
||||
// double-quoting is an injection/correctness risk. Single-quoting (with the embedded
|
||||
// `'` → `'\''` escape) is literal and safe for slashes (e.g. "release/2026-06") too.
|
||||
const shellQuote = (s: string): string => `'${s.replace(/'/g, "'\\''")}'`;
|
||||
const localRef = shellQuote(branch);
|
||||
const originRef = shellQuote(`origin/${branch}`);
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-09:00:
|
||||
Shell-quote with a real single-quoted POSIX literal, NOT JSON.stringify. A
|
||||
JSON double-quoted string still lets bash expand `$(...)`, backticks, and `$VAR`
|
||||
inside it; JSON.stringify is not a shell-quoting function. Git ref names can't
|
||||
legally contain backticks so there's no live injection path today, but
|
||||
single-quoting is the idiomatic safe form and stays correct if a caller ever
|
||||
passes a less-constrained string. A single quote inside the value is escaped as
|
||||
the standard `'\''` close-reopen sequence.
|
||||
*/
|
||||
const shellSingleQuote = (value: string): string => `'${value.replace(/'/g, "'\\''")}'`;
|
||||
const localRef = shellSingleQuote(branch);
|
||||
const originRef = shellSingleQuote(`origin/${branch}`);
|
||||
let baseCommitSha: string | undefined;
|
||||
try {
|
||||
const { stdout } = await execAsync(
|
||||
|
||||
@@ -9175,6 +9175,11 @@ export class TaskExecutor {
|
||||
// mirroring the primary execute-seam session above.
|
||||
actionGateContext: this.buildActionGateContext(task.id, identityAgent, settings.defaultAgentPermissionPolicy),
|
||||
permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, identityAgent, settings.defaultAgentPermissionPolicy),
|
||||
// FNXC:SessionRouting 2026-06-24-11:20:
|
||||
// #1675: propagate task id so retry-session requests carry the same
|
||||
// X-Session-Id/X-Session-Affinity as the primary session, keeping the
|
||||
// task's LLM requests grouped under one stable routing/observability id.
|
||||
taskId: task.id,
|
||||
});
|
||||
retrySession = createdRetrySession.session;
|
||||
if (createdRetrySession.sessionFile) {
|
||||
@@ -12238,6 +12243,10 @@ Do not refactor, rename broadly, or make opportunistic improvements.
|
||||
runAuditor: createRunAuditor(this.store, this.getRunContextFor(task.id)),
|
||||
settings,
|
||||
taskEnv: extraEnv,
|
||||
// FNXC:SessionRouting 2026-06-24-11:20:
|
||||
// #1675: propagate task id so verification-fix requests carry the same
|
||||
// X-Session-Id/X-Session-Affinity as the primary session.
|
||||
taskId: task.id,
|
||||
...(skillContext?.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}),
|
||||
});
|
||||
|
||||
@@ -13534,6 +13543,10 @@ You have access to the file system to review changes.${verdictBlock}`;
|
||||
runAuditor: createRunAuditor(this.store, this.getRunContextFor(task.id)),
|
||||
settings,
|
||||
taskEnv: stepEnv,
|
||||
// FNXC:SessionRouting 2026-06-24-11:20:
|
||||
// #1675: propagate task id so workflow-step requests carry the same
|
||||
// X-Session-Id/X-Session-Affinity as the primary session.
|
||||
taskId: task.id,
|
||||
// Skill selection: assigned-agent / role-fallback skills, plus the step's
|
||||
// own named skill (U1) made discoverable via additionalSkillPaths.
|
||||
...(effectiveSkillSelection ? { skillSelection: effectiveSkillSelection } : {}),
|
||||
@@ -15093,10 +15106,18 @@ You have access to the file system to review changes.${verdictBlock}`;
|
||||
const tasks = await this.store.listTasks({ slim: true, includeArchived: false });
|
||||
for (const t of tasks) {
|
||||
if (t.id === requestingTaskId) continue;
|
||||
if (t.worktree !== worktreePath) continue;
|
||||
if (t.column !== "in-progress") continue;
|
||||
if (t.paused === true) continue;
|
||||
return t.id;
|
||||
if (t.worktree === worktreePath) return t.id;
|
||||
// FNXC:Workspace 2026-06-22-09:00: workspace tasks hold their worktrees in
|
||||
// task.workspaceWorktrees, not the singular task.worktree column. The DB liveness
|
||||
// fallback must check those per-sub-repo paths too — otherwise a conflict against a
|
||||
// sub-repo worktree owned by an in-progress workspace task is missed, especially
|
||||
// before its in-memory activeWorktrees entry is (re)registered after restart.
|
||||
const wsEntries = t.workspaceWorktrees;
|
||||
if (wsEntries && Object.values(wsEntries).some((entry) => entry.worktreePath === worktreePath)) {
|
||||
return t.id;
|
||||
}
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
@@ -16287,6 +16308,10 @@ Child agent: ${agent.id} (${name})`;
|
||||
runAuditor: createRunAuditor(this.store, this.getRunContextFor(taskId)),
|
||||
settings,
|
||||
taskEnv,
|
||||
// FNXC:SessionRouting 2026-06-24-11:20:
|
||||
// #1675: propagate task id so child-agent requests carry the same
|
||||
// X-Session-Id/X-Session-Affinity as the parent task session.
|
||||
taskId,
|
||||
// Skill selection: use assigned agent skills if available, otherwise role fallback
|
||||
...(skillContext.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}),
|
||||
});
|
||||
|
||||
@@ -1895,6 +1895,64 @@ export function wrapToolsWithActionGate(
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:SessionRouting 2026-06-23-16:40:
|
||||
* Outbound LLM chat completion requests must carry `X-Session-Id` and
|
||||
* `X-Session-Affinity` headers (GitHub issue #1675). These are widely
|
||||
* understood by LLM gateways, proxies, and observability tooling:
|
||||
* - Gateways/routers use them for sticky routing, keeping consecutive requests
|
||||
* from one conversation on the same backend or cache instance.
|
||||
* - Observability tools (e.g. Langfuse, Arize) use them to group individually
|
||||
* stateless API calls into a single cohesive multi-turn chat trace.
|
||||
* - Memory/proxy middleware uses them to fetch and append conversation history.
|
||||
*
|
||||
* Both headers carry the same stable identifier so sticky-routing affinity and
|
||||
* trace grouping refer to the same session. Builds the header pair for a given
|
||||
* session id.
|
||||
*/
|
||||
export function buildSessionRoutingHeaders(sessionId: string): Record<string, string> {
|
||||
return {
|
||||
"X-Session-Id": sessionId,
|
||||
"X-Session-Affinity": sessionId,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:SessionRouting 2026-06-23-16:40:
|
||||
* Merge the session-routing headers into every header set the model registry
|
||||
* resolves for outbound LLM requests (#1675). `getApiKeyAndHeaders` is the
|
||||
* single point pi-coding-agent uses to resolve per-request auth and headers
|
||||
* (for the main stream and compaction alike), so wrapping it applies the
|
||||
* headers to every HTTP-based provider path (built-in, custom, and
|
||||
* HTTP-streaming extension providers). Subprocess-based providers that make
|
||||
* their own outbound HTTP calls inside a child process (e.g. CLI bridges) are
|
||||
* outside this seam and do not inherit the headers.
|
||||
* Operating on the resolved output (rather than re-registering providers)
|
||||
* preserves provider-specific headers and never disturbs API-key resolution.
|
||||
*/
|
||||
export function attachSessionRoutingHeaders(modelRegistry: ModelRegistry, sessionId: string): void {
|
||||
// FNXC:SessionRouting 2026-06-23-16:46:
|
||||
// Auxiliary feature: never let header injection break session creation. If a
|
||||
// future pi-coding-agent rename removes getApiKeyAndHeaders, warn (rather than
|
||||
// silently no-op) so the degraded routing/observability headers are detectable.
|
||||
if (typeof modelRegistry.getApiKeyAndHeaders !== "function") {
|
||||
piLog.warn("[pi] session-routing headers not attached: ModelRegistry.getApiKeyAndHeaders is not a function (pi API changed?)");
|
||||
return;
|
||||
}
|
||||
const routingHeaders = buildSessionRoutingHeaders(sessionId);
|
||||
const resolveAuth = modelRegistry.getApiKeyAndHeaders.bind(modelRegistry);
|
||||
modelRegistry.getApiKeyAndHeaders = async (model) => {
|
||||
const result = await resolveAuth(model);
|
||||
if (!result.ok) {
|
||||
return result;
|
||||
}
|
||||
return {
|
||||
...result,
|
||||
headers: { ...result.headers, ...routingHeaders },
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a pi agent session configured for fn.
|
||||
* Reuses the user's existing pi auth and model configuration.
|
||||
@@ -2098,6 +2156,20 @@ export async function createFnAgent(options: AgentOptions): Promise<AgentResult>
|
||||
const sessionManager = options.sessionManager ?? SessionManager.inMemory();
|
||||
normalizeSessionHistoryEntries(sessionManager as unknown as SessionManagerLike);
|
||||
|
||||
// FNXC:SessionRouting 2026-06-23-16:40:
|
||||
// Tag every outbound LLM chat completion request with stable session-routing
|
||||
// headers (X-Session-Id / X-Session-Affinity) for gateway sticky routing and
|
||||
// observability trace grouping (#1675). Prefer the task id, which is stable
|
||||
// across pause/resume (each resume spins up a fresh SessionManager), and fall
|
||||
// back to the pi session id for non-task sessions (chat, summarizer, reviewer).
|
||||
const piSessionId = typeof sessionManager.getSessionId === "function"
|
||||
? sessionManager.getSessionId()
|
||||
: undefined;
|
||||
const sessionRoutingId = options.taskId ?? piSessionId;
|
||||
if (sessionRoutingId) {
|
||||
attachSessionRoutingHeaders(modelRegistry, sessionRoutingId);
|
||||
}
|
||||
|
||||
const createSessionWithModel = async (modelOverride?: typeof selectedModel) => {
|
||||
// pi-coding-agent 0.68+: `tools` is a string[] allowlist of tool names, not
|
||||
// Tool instances. We need boundary-wrapped versions of the built-ins, so we
|
||||
|
||||
@@ -744,6 +744,24 @@ export async function acquireWorkspaceRepoWorktree(
|
||||
await store.logEntry(task.id, `Remembered workspace worktree for ${repoRelPath} is no longer usable; re-acquiring`, existing.worktreePath, runContext);
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-09:00:
|
||||
Run best-effort observability (task log + audit) for the NON-FATAL post-acquire
|
||||
steps without letting their own awaited writes escape. logEntry/audit can throw
|
||||
(DB hiccup, audit sink failure); an unsuppressed throw inside a non-fatal catch
|
||||
would re-escalate guard/base-capture failures into fatal acquisition errors that
|
||||
strand the already-created worktree. Mirrors the busy-path swallow above.
|
||||
*/
|
||||
const safeObserve = async (fn: () => Promise<void>): Promise<void> => {
|
||||
try {
|
||||
await fn();
|
||||
} catch (obsErr) {
|
||||
logger?.warn(
|
||||
`${task.id}: workspace acquisition observability failed (suppressed): ${obsErr instanceof Error ? obsErr.message : String(obsErr)}`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-21-20:10:
|
||||
Same-sub-repo exclusivity (KTD4): register the sub-repo absolute path in the
|
||||
@@ -854,16 +872,18 @@ export async function acquireWorkspaceRepoWorktree(
|
||||
// Swallow logging errors so acquisition continues (matching the F6 busy-path defensive wrap above).
|
||||
const message = guardErr instanceof Error ? guardErr.message : String(guardErr);
|
||||
logger?.warn(`${task.id}: identity-guard install failed for sub-repo ${repoRelPath} (non-fatal): ${message}`);
|
||||
try {
|
||||
// FNXC:Workspace 2026-06-22-09:00: the observability writes (store.logEntry / audit.git)
|
||||
// are themselves awaited and can throw; an unwrapped throw here would escape the catch
|
||||
// and re-escalate this deliberately NON-FATAL step into a fatal acquisition error,
|
||||
// stranding the already-created worktree. Suppress observability failures via safeObserve.
|
||||
await safeObserve(async () => {
|
||||
await store.logEntry(task.id, `Workspace sub-repo identity-guard install failed for ${repoRelPath} (non-fatal): ${message}`, undefined, runContext);
|
||||
await audit?.git({
|
||||
type: "worktree:workspace-repo-acquire-failed",
|
||||
target: repoAbsPath,
|
||||
metadata: { repoRelPath, taskId: task.id, error: message, stage: "identity-guard" },
|
||||
});
|
||||
} catch {
|
||||
// best-effort observability only — keep the (non-fatal) guard failure non-fatal
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -891,16 +911,16 @@ export async function acquireWorkspaceRepoWorktree(
|
||||
// non-fatal capture failure into a fatal acquisition failure (parity with the F6 busy-path defensive wrap).
|
||||
const message = baseErr instanceof Error ? baseErr.message : String(baseErr);
|
||||
logger?.warn(`${task.id}: base-SHA capture failed for sub-repo ${repoRelPath} (non-fatal): ${message}`);
|
||||
try {
|
||||
// FNXC:Workspace 2026-06-22-09:00: same non-fatal contract as the identity-guard catch —
|
||||
// the awaited observability writes must not re-escalate a non-fatal base-capture failure.
|
||||
await safeObserve(async () => {
|
||||
await store.logEntry(task.id, `Workspace sub-repo base-SHA capture failed for ${repoRelPath} (non-fatal): ${message}`, undefined, runContext);
|
||||
await audit?.git({
|
||||
type: "worktree:workspace-repo-acquire-failed",
|
||||
target: repoAbsPath,
|
||||
metadata: { repoRelPath, taskId: task.id, error: message, stage: "base-sha-capture" },
|
||||
});
|
||||
} catch {
|
||||
// best-effort observability only — keep the (non-fatal) capture failure non-fatal
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -918,7 +938,19 @@ export async function acquireWorkspaceRepoWorktree(
|
||||
...(latest.workspaceWorktrees ?? {}),
|
||||
[repoRelPath]: { worktreePath: result.worktreePath, branch: result.branch, baseCommitSha },
|
||||
};
|
||||
await store.updateTask(task.id, { workspaceWorktrees: updated });
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-09:00:
|
||||
F10 — reset the singular worktree/branch columns to null in the SAME write that
|
||||
persists workspaceWorktrees. The single-repo `acquireTaskWorktree` above wrote
|
||||
`task.worktree`/`task.branch` (the sub-repo path/branch) to the real task row;
|
||||
clearing the in-memory copy passed in only stops the NEXT sub-repo from resuming
|
||||
into this one's worktree — the DB row stays polluted. A non-null `task.worktree`
|
||||
makes `isWorkspaceTask(task)` return false (its first guard), so the dashboard
|
||||
stops rendering WorkspaceWorktreesSummary and instead shows the sub-repo branch in
|
||||
the standard chip — the blank/wrong-card state U10 prevents. Nulling them here
|
||||
keeps `task.worktree` null for the workspace task's whole lifetime.
|
||||
*/
|
||||
await store.updateTask(task.id, { workspaceWorktrees: updated, worktree: null, branch: null });
|
||||
|
||||
return { worktreePath: result.worktreePath, branch: result.branch, baseCommitSha, alreadyAcquired: false };
|
||||
} catch (err) {
|
||||
@@ -935,16 +967,19 @@ export async function acquireWorkspaceRepoWorktree(
|
||||
// surface a logging error instead. Best-effort observability; `err` is always re-thrown.
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
logger?.error?.(`${task.id}: workspace sub-repo acquisition failed for ${repoRelPath}: ${message}`);
|
||||
try {
|
||||
// FNXC:Workspace 2026-06-22-09:30: the fatal-path observability writes must use safeObserve
|
||||
// for the same reason as the non-fatal catches — an unsuppressed throw from logEntry/audit
|
||||
// would replace `err` as the propagated rejection, so a store/audit hiccup could surface a
|
||||
// non-WorkspaceRepoAcquireBusyError to callers whose `instanceof` type checks then misfire.
|
||||
// The original acquisition `err` (line below) is the contract; observability is best-effort.
|
||||
await safeObserve(async () => {
|
||||
await store.logEntry(task.id, `Workspace sub-repo acquisition failed for ${repoRelPath}: ${message}`, undefined, runContext);
|
||||
await audit?.git({
|
||||
type: "worktree:workspace-repo-acquire-failed",
|
||||
target: repoAbsPath,
|
||||
metadata: { repoRelPath, taskId: task.id, error: message },
|
||||
});
|
||||
} catch {
|
||||
// best-effort observability only — ensure the original acquisition error propagates
|
||||
}
|
||||
});
|
||||
}
|
||||
throw err;
|
||||
} finally {
|
||||
|
||||
Reference in New Issue
Block a user