feat(server): configure shared workspace concurrency (#10759)
<!-- Write all pull request text in Simplified Technical English (ASD-STE100): short sentences, one instruction per sentence, simple approved vocabulary, and the active voice. --> ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The heartbeat service starts agent runs in local, SSH, sandbox, plugin, or Kubernetes environments. > - Runs can share one project working tree. > - PR #10699 made every shared working tree single-file, including trusted local and SSH hosts that support coordinated concurrent work. > - Operators need a policy that keeps remote environments safe and restores local multi-agent work. > - This pull request adds an `auto`, `serialize`, or `allow` concurrency policy and applies it to the final execution environment. > - The benefit is safe serialization by default for sandboxed targets and useful concurrency by default for persistent local targets. ## Linked Issues or Issue Description Related work: #10699 introduced the busy gate that this change makes configurable. #7852 covers a separate environment-lease race and does not provide this dispatch policy. **Subsystem affected** `server/` heartbeat orchestration and `packages/shared/` workspace policy contracts. **Problem or motivation** The shared-workspace busy gate always defers a second run. This behavior prevents local multi-agent projects from running concurrently even when operators expect agents to coordinate through commits. **Proposed solution** Add `sharedWorkspaceConcurrency` with `auto`, `serialize`, and `allow` values. Default `auto` permits local and SSH concurrency. It serializes sandbox, plugin, and forced Kubernetes execution. **Alternatives considered** Keeping unconditional serialization is too restrictive for persistent host working trees. Always allowing overlap removes the protection that sandboxed and remote targets need. **Roadmap alignment** This is a focused correction to the shipped cloud and sandbox execution milestone. It does not add a new roadmap feature. ## What Changed - Added the optional tri-state field to project policy and issue override contracts and validators. - Added a pure resolver with issue override, project policy, and `auto` default precedence. - Moved final environment and Kubernetes resolution before the shared-workspace busy gate. - Kept the existing deferral and retry behavior for every path that resolves to serialization. - Added a task-context warning and a structured log when a run dispatches beside a live holder. - Added policy and heartbeat coverage for all requested policy and environment combinations. - Documented the new policy and its default behavior. ## Verification - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/execution-workspace-policy.test.ts src/__tests__/heartbeat-workspace-busy.test.ts` — 32 tests passed. - `pnpm -r typecheck` — passed. - `PAPERCLIP_IN_WORKTREE=false PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS=false PAPERCLIP_RESTORE_IN_PROGRESS=false pnpm test:run` — all 3,528 server tests passed. The later UI/workspace phase passed 3,384 tests and hit one unrelated `CompanyEnvironments` navigation timing failure. - `pnpm --filter @paperclipai/ui exec vitest run src/pages/CompanyEnvironments.test.tsx -t "opens the edit form on a standalone page with existing values and closes after save"` — the unrelated UI test passed in isolation. - `pnpm build` — passed. Policy matrix: | Policy | Final target | Expected result | Result | | --- | --- | --- | --- | | `auto` | local | Dispatch with holder note | Passed | | `auto` | sandbox | Defer with `workspace_busy` | Passed | | `auto` | instance-forced Kubernetes | Defer with `workspace_busy` | Passed | | `serialize` | local | Defer with `workspace_busy` | Passed | | `allow` | sandbox | Dispatch with holder note | Passed | Existing serialization, retry, and stale-holder tests also pass. ## Risks - `auto` changes the post-#10699 local and SSH behavior back to concurrent dispatch. Concurrent agents can mutate the same working tree, so each dispatched run receives an explicit coordination warning. - `allow` is an operator override and can permit overlap in sandbox or plugin environments. - Unknown environment drivers serialize in `auto` mode. This keeps the fallback conservative. - There is no database migration. An absent field resolves to `auto`. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI GPT-5 (Codex). The exact API snapshot and context-window size are not exposed to the agent. Reasoning, repository editing, terminal tool use, and local code execution were enabled. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
42d0ddcb86
commit
ba396c608c
|
|
@ -22,6 +22,16 @@ We are intentionally not shipping the UI for this yet. The runtime code remains
|
|||
- seeded worktree instances can keep local-encrypted secrets working
|
||||
- seeded worktree instances can rebind same-repo project workspace paths onto the current git worktree
|
||||
|
||||
## Shared workspace concurrency policy
|
||||
|
||||
Projects and individual issues can set `sharedWorkspaceConcurrency` in their execution workspace policy/settings:
|
||||
|
||||
- `auto` (the default when absent): allow concurrent shared-workspace runs on `local` and `ssh` environments, and serialize runs on `sandbox` and `plugin` environments. An instance forced to Kubernetes always serializes in `auto` mode.
|
||||
- `serialize`: defer a run while another live run holds the same project workspace, using the `workspace_busy` retry path.
|
||||
- `allow`: dispatch alongside a live holder on every environment.
|
||||
|
||||
Issue settings override the project policy, which overrides the default `auto`. When concurrency is allowed and a live holder exists, Paperclip adds the holder run and issue to the dispatched task context so agents can coordinate concurrent mutations through commits. The setting is optional JSON policy data, so existing databases require no migration.
|
||||
|
||||
## Hidden UI entrypoints
|
||||
|
||||
These are the current user-facing UI surfaces for the feature, now intentionally disabled:
|
||||
|
|
|
|||
|
|
@ -845,6 +845,7 @@ export type {
|
|||
WorkspaceRealizationTransport,
|
||||
ExecutionWorkspaceStrategyType,
|
||||
ExecutionWorkspaceMode,
|
||||
SharedWorkspaceConcurrency,
|
||||
ExecutionWorkspaceProviderType,
|
||||
ExecutionWorkspaceStatus,
|
||||
ExecutionWorkspaceStrategy,
|
||||
|
|
|
|||
|
|
@ -361,6 +361,7 @@ export type {
|
|||
WorkspaceRealizationTransport,
|
||||
ExecutionWorkspaceStrategyType,
|
||||
ExecutionWorkspaceMode,
|
||||
SharedWorkspaceConcurrency,
|
||||
ExecutionWorkspaceProviderType,
|
||||
ExecutionWorkspaceStatus,
|
||||
ExecutionWorkspaceStrategy,
|
||||
|
|
|
|||
|
|
@ -20,6 +20,8 @@ export type ExecutionWorkspaceMode =
|
|||
| "reuse_existing"
|
||||
| "agent_default";
|
||||
|
||||
export type SharedWorkspaceConcurrency = "auto" | "serialize" | "allow";
|
||||
|
||||
export type ExecutionWorkspaceProviderType =
|
||||
| "local_fs"
|
||||
| "git_worktree"
|
||||
|
|
@ -149,6 +151,7 @@ export interface ExecutionWorkspaceCloseReadiness {
|
|||
|
||||
export interface ProjectExecutionWorkspacePolicy {
|
||||
enabled: boolean;
|
||||
sharedWorkspaceConcurrency?: SharedWorkspaceConcurrency;
|
||||
defaultMode?: ProjectExecutionWorkspaceDefaultMode;
|
||||
allowIssueOverride?: boolean;
|
||||
defaultProjectWorkspaceId?: string | null;
|
||||
|
|
@ -164,6 +167,7 @@ export interface ProjectExecutionWorkspacePolicy {
|
|||
|
||||
export interface IssueExecutionWorkspaceSettings {
|
||||
mode?: ExecutionWorkspaceMode;
|
||||
sharedWorkspaceConcurrency?: SharedWorkspaceConcurrency;
|
||||
environmentId?: string | null;
|
||||
workspaceStrategy?: ExecutionWorkspaceStrategy | null;
|
||||
workspaceRuntime?: Record<string, unknown> | null;
|
||||
|
|
|
|||
|
|
@ -152,6 +152,7 @@ function isAllowedTaskEgressCidr(cidr: string): boolean {
|
|||
export const issueExecutionWorkspaceSettingsSchema = z
|
||||
.object({
|
||||
mode: z.enum(ISSUE_EXECUTION_WORKSPACE_PREFERENCES).optional(),
|
||||
sharedWorkspaceConcurrency: z.enum(["auto", "serialize", "allow"]).optional(),
|
||||
environmentId: z.string().uuid().optional().nullable(),
|
||||
workspaceStrategy: executionWorkspaceStrategySchema.optional().nullable(),
|
||||
workspaceRuntime: z.record(z.string(), z.unknown()).optional().nullable(),
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ const executionWorkspaceStrategySchema = z
|
|||
export const projectExecutionWorkspacePolicySchema = z
|
||||
.object({
|
||||
enabled: z.boolean(),
|
||||
sharedWorkspaceConcurrency: z.enum(["auto", "serialize", "allow"]).optional(),
|
||||
defaultMode: z.enum(["shared_workspace", "isolated_workspace", "operator_branch", "adapter_default"]).optional(),
|
||||
allowIssueOverride: z.boolean().optional(),
|
||||
defaultProjectWorkspaceId: z.string().uuid().optional().nullable(),
|
||||
|
|
|
|||
|
|
@ -1,4 +1,8 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
issueExecutionWorkspaceSettingsSchema,
|
||||
projectExecutionWorkspacePolicySchema,
|
||||
} from "@paperclipai/shared";
|
||||
import {
|
||||
buildExecutionWorkspaceAdapterConfig,
|
||||
defaultIssueExecutionWorkspaceSettingsForProject,
|
||||
|
|
@ -10,6 +14,7 @@ import {
|
|||
resolveExecutionWorkspaceEnvironmentId,
|
||||
resolvePinnedIssueWorkspaceStrategyType,
|
||||
resolveExecutionWorkspaceMode,
|
||||
resolveSharedWorkspaceConcurrency,
|
||||
selectEnvironmentExecutionWorkspaceSettings,
|
||||
} from "../services/execution-workspace-policy.ts";
|
||||
|
||||
|
|
@ -40,6 +45,42 @@ describe("execution workspace policy helpers", () => {
|
|||
).toBe("isolated_workspace");
|
||||
});
|
||||
|
||||
it("resolves shared-workspace concurrency from issue override, project policy, then auto", () => {
|
||||
expect(
|
||||
resolveSharedWorkspaceConcurrency({
|
||||
projectPolicy: { enabled: true, sharedWorkspaceConcurrency: "serialize" },
|
||||
issueSettings: { sharedWorkspaceConcurrency: "allow" },
|
||||
}),
|
||||
).toBe("allow");
|
||||
expect(
|
||||
resolveSharedWorkspaceConcurrency({
|
||||
projectPolicy: { enabled: true, sharedWorkspaceConcurrency: "serialize" },
|
||||
issueSettings: null,
|
||||
}),
|
||||
).toBe("serialize");
|
||||
expect(
|
||||
resolveSharedWorkspaceConcurrency({
|
||||
projectPolicy: { enabled: false, sharedWorkspaceConcurrency: "serialize" },
|
||||
issueSettings: null,
|
||||
}),
|
||||
).toBe("auto");
|
||||
expect(resolveSharedWorkspaceConcurrency({ projectPolicy: null, issueSettings: null })).toBe("auto");
|
||||
});
|
||||
|
||||
it("validates the shared-workspace concurrency enum on project and issue settings", () => {
|
||||
expect(projectExecutionWorkspacePolicySchema.parse({
|
||||
enabled: true,
|
||||
sharedWorkspaceConcurrency: "auto",
|
||||
}).sharedWorkspaceConcurrency).toBe("auto");
|
||||
expect(issueExecutionWorkspaceSettingsSchema.parse({
|
||||
sharedWorkspaceConcurrency: "allow",
|
||||
}).sharedWorkspaceConcurrency).toBe("allow");
|
||||
expect(projectExecutionWorkspacePolicySchema.safeParse({
|
||||
enabled: true,
|
||||
sharedWorkspaceConcurrency: "parallel",
|
||||
}).success).toBe(false);
|
||||
});
|
||||
|
||||
it("centralizes unrunnable isolated worktree detection", () => {
|
||||
expect(
|
||||
isUnrunnableWorktreeCombo({
|
||||
|
|
@ -256,6 +297,7 @@ describe("execution workspace policy helpers", () => {
|
|||
expect(
|
||||
parseProjectExecutionWorkspacePolicy({
|
||||
enabled: true,
|
||||
sharedWorkspaceConcurrency: "serialize",
|
||||
defaultMode: "isolated",
|
||||
workspaceStrategy: {
|
||||
type: "git_worktree",
|
||||
|
|
@ -267,6 +309,7 @@ describe("execution workspace policy helpers", () => {
|
|||
}),
|
||||
).toEqual({
|
||||
enabled: true,
|
||||
sharedWorkspaceConcurrency: "serialize",
|
||||
defaultMode: "isolated_workspace",
|
||||
workspaceStrategy: {
|
||||
type: "git_worktree",
|
||||
|
|
@ -299,6 +342,7 @@ describe("execution workspace policy helpers", () => {
|
|||
expect(
|
||||
parseIssueExecutionWorkspaceSettings({
|
||||
mode: "isolated_workspace",
|
||||
sharedWorkspaceConcurrency: "allow",
|
||||
networkEgress: {
|
||||
allowFqdns: ["github.com", "pypi.org"],
|
||||
allowCidrs: ["203.0.113.0/24"],
|
||||
|
|
@ -306,6 +350,7 @@ describe("execution workspace policy helpers", () => {
|
|||
}),
|
||||
).toEqual({
|
||||
mode: "isolated_workspace",
|
||||
sharedWorkspaceConcurrency: "allow",
|
||||
networkEgress: {
|
||||
allowFqdns: ["github.com", "pypi.org"],
|
||||
allowCidrs: ["203.0.113.0/24"],
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ import {
|
|||
companies,
|
||||
companySkills,
|
||||
createDb,
|
||||
environments,
|
||||
environmentLeases,
|
||||
executionWorkspaces,
|
||||
heartbeatRunEvents,
|
||||
|
|
@ -28,7 +29,11 @@ import {
|
|||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js";
|
||||
import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts";
|
||||
import {
|
||||
registerServerAdapter,
|
||||
unregisterServerAdapter,
|
||||
type AdapterExecutionContext,
|
||||
} from "../adapters/index.ts";
|
||||
import {
|
||||
WORKSPACE_BUSY_ERROR_CODE,
|
||||
WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS,
|
||||
|
|
@ -76,6 +81,7 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let workspaceCwd!: string;
|
||||
const executedRunIds: string[] = [];
|
||||
const executedInputs = new Map<string, AdapterExecutionContext>();
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-workspace-busy-");
|
||||
|
|
@ -84,8 +90,9 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
workspaceCwd = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-busy-"));
|
||||
registerServerAdapter({
|
||||
type: WORKSPACE_BUSY_TEST_ADAPTER,
|
||||
execute: async (input: { runId?: string }) => {
|
||||
executedRunIds.push(input.runId ?? "unknown");
|
||||
execute: async (input) => {
|
||||
executedRunIds.push(input.runId);
|
||||
executedInputs.set(input.runId, input);
|
||||
return {
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
|
|
@ -113,6 +120,8 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
await cleanupFixture();
|
||||
executedRunIds.length = 0;
|
||||
executedInputs.clear();
|
||||
await instanceSettingsService(db).updateGeneral({ executionMode: "any" });
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
|
|
@ -152,6 +161,7 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
await db.delete(agentRuntimeState);
|
||||
await db.delete(budgetPolicies);
|
||||
await db.delete(agents);
|
||||
await db.delete(environments);
|
||||
await db.delete(companySkills);
|
||||
await db.delete(companies);
|
||||
}
|
||||
|
|
@ -182,7 +192,10 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
holderIssueWorkspaceSettings?: Record<string, unknown> | null;
|
||||
holderProjectWorkspaceId?: string;
|
||||
holderActivityAt?: Date;
|
||||
issueWorkspaceSettings?: Record<string, unknown> | null;
|
||||
agentEnvironmentDriver?: "sandbox";
|
||||
}): Promise<WorkspaceFixture> {
|
||||
await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: true });
|
||||
const companyId = randomUUID();
|
||||
const projectId = randomUUID();
|
||||
const projectWorkspaceId = randomUUID();
|
||||
|
|
@ -192,6 +205,7 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
const agentId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const nonAssigneeAgentId = randomUUID();
|
||||
const agentEnvironmentId = input?.agentEnvironmentDriver ? randomUUID() : null;
|
||||
const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`;
|
||||
const now = new Date();
|
||||
|
||||
|
|
@ -219,6 +233,17 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
isPrimary: true,
|
||||
});
|
||||
|
||||
if (agentEnvironmentId) {
|
||||
await db.insert(environments).values({
|
||||
id: agentEnvironmentId,
|
||||
companyId,
|
||||
name: `Workspace busy ${input!.agentEnvironmentDriver} ${agentEnvironmentId}`,
|
||||
driver: input!.agentEnvironmentDriver!,
|
||||
status: "active",
|
||||
config: { provider: "fake", image: "fake:test", reuseLease: false },
|
||||
});
|
||||
}
|
||||
|
||||
const holderProjectWorkspaceId = input?.holderProjectWorkspaceId ?? projectWorkspaceId;
|
||||
if (holderProjectWorkspaceId !== projectWorkspaceId) {
|
||||
await db.insert(projectWorkspaces).values({
|
||||
|
|
@ -250,6 +275,7 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
maxConcurrentRuns: 1,
|
||||
},
|
||||
},
|
||||
...(id === agentId && agentEnvironmentId ? { defaultEnvironmentId: agentEnvironmentId } : {}),
|
||||
permissions: {},
|
||||
});
|
||||
}
|
||||
|
|
@ -304,6 +330,10 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
projectWorkspaceId,
|
||||
issueNumber: 2,
|
||||
identifier: `${issuePrefix}-2`,
|
||||
executionWorkspaceSettings:
|
||||
input?.issueWorkspaceSettings === undefined
|
||||
? { sharedWorkspaceConcurrency: "serialize" }
|
||||
: input.issueWorkspaceSettings,
|
||||
});
|
||||
|
||||
return {
|
||||
|
|
@ -319,6 +349,122 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
|||
};
|
||||
}
|
||||
|
||||
it("auto dispatches alongside a local holder and adds coordination context", async () => {
|
||||
const fixture = await seedWorkspaceFixture({
|
||||
issueWorkspaceSettings: { sharedWorkspaceConcurrency: "auto" },
|
||||
});
|
||||
|
||||
const run = await heartbeat.invoke(
|
||||
fixture.agentId,
|
||||
"assignment",
|
||||
{ issueId: fixture.issueId, wakeReason: "issue_assigned" },
|
||||
"system",
|
||||
);
|
||||
expect(run).not.toBeNull();
|
||||
|
||||
const finishedRun = await waitForRunToLeaveActiveStates(run!.id);
|
||||
expect(finishedRun?.status).toBe("succeeded");
|
||||
expect(executedRunIds).toContain(run!.id);
|
||||
expect(executedInputs.get(run!.id)?.context.paperclipTaskMarkdown).toContain(
|
||||
`shared workspace is concurrently held by run ${fixture.holderRunId}`,
|
||||
);
|
||||
expect(executedInputs.get(run!.id)?.context.paperclipTaskMarkdown).toContain(
|
||||
"expect concurrent mutations, coordinate via commits",
|
||||
);
|
||||
});
|
||||
|
||||
it("auto defers when the final environment driver is sandbox", async () => {
|
||||
const fixture = await seedWorkspaceFixture({
|
||||
issueWorkspaceSettings: { sharedWorkspaceConcurrency: "auto" },
|
||||
agentEnvironmentDriver: "sandbox",
|
||||
});
|
||||
|
||||
const run = await heartbeat.invoke(
|
||||
fixture.agentId,
|
||||
"assignment",
|
||||
{ issueId: fixture.issueId, wakeReason: "issue_assigned" },
|
||||
"system",
|
||||
);
|
||||
expect(run).not.toBeNull();
|
||||
|
||||
const finishedRun = await waitForRunToLeaveActiveStates(run!.id);
|
||||
expect(finishedRun?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE);
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
});
|
||||
|
||||
it("auto defers when instance policy forces Kubernetes", async () => {
|
||||
const fixture = await seedWorkspaceFixture({
|
||||
issueWorkspaceSettings: { sharedWorkspaceConcurrency: "auto" },
|
||||
});
|
||||
await db.insert(environments).values({
|
||||
id: randomUUID(),
|
||||
companyId: fixture.companyId,
|
||||
name: `Managed Kubernetes ${fixture.companyId}`,
|
||||
driver: "sandbox",
|
||||
status: "active",
|
||||
config: { provider: "kubernetes" },
|
||||
metadata: { managedKubernetesSandbox: true },
|
||||
});
|
||||
await instanceSettingsService(db).updateGeneral({ executionMode: "kubernetes" });
|
||||
|
||||
const run = await heartbeat.invoke(
|
||||
fixture.agentId,
|
||||
"assignment",
|
||||
{ issueId: fixture.issueId, wakeReason: "issue_assigned" },
|
||||
"system",
|
||||
);
|
||||
expect(run).not.toBeNull();
|
||||
|
||||
const finishedRun = await waitForRunToLeaveActiveStates(run!.id);
|
||||
expect(finishedRun?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE);
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
});
|
||||
|
||||
it("serialize defers even when the final environment driver is local", async () => {
|
||||
const fixture = await seedWorkspaceFixture({
|
||||
issueWorkspaceSettings: { sharedWorkspaceConcurrency: "serialize" },
|
||||
});
|
||||
|
||||
const run = await heartbeat.invoke(
|
||||
fixture.agentId,
|
||||
"assignment",
|
||||
{ issueId: fixture.issueId, wakeReason: "issue_assigned" },
|
||||
"system",
|
||||
);
|
||||
expect(run).not.toBeNull();
|
||||
|
||||
const finishedRun = await waitForRunToLeaveActiveStates(run!.id);
|
||||
expect(finishedRun?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE);
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
});
|
||||
|
||||
it("allow passes the busy gate for a sandbox environment and adds coordination context", async () => {
|
||||
const fixture = await seedWorkspaceFixture({
|
||||
issueWorkspaceSettings: { sharedWorkspaceConcurrency: "allow" },
|
||||
agentEnvironmentDriver: "sandbox",
|
||||
});
|
||||
|
||||
const run = await heartbeat.invoke(
|
||||
fixture.agentId,
|
||||
"assignment",
|
||||
{ issueId: fixture.issueId, wakeReason: "issue_assigned" },
|
||||
"system",
|
||||
);
|
||||
expect(run).not.toBeNull();
|
||||
|
||||
const finishedRun = await waitForRunToLeaveActiveStates(run!.id);
|
||||
expect(finishedRun?.errorCode).not.toBe(WORKSPACE_BUSY_ERROR_CODE);
|
||||
expect(executedRunIds).toContain(run!.id);
|
||||
expect((finishedRun?.contextSnapshot as Record<string, unknown>)?.paperclipTaskMarkdown).toContain(
|
||||
`shared workspace is concurrently held by run ${fixture.holderRunId}`,
|
||||
);
|
||||
const retryRuns = await db
|
||||
.select({ id: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.scheduledRetryReason, WORKSPACE_BUSY_RETRY_REASON));
|
||||
expect(retryRuns).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("defers a run whose issue targets a busy shared workspace and schedules a bounded retry", async () => {
|
||||
const fixture = await seedWorkspaceFixture();
|
||||
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import type {
|
|||
IssueExecutionWorkspaceSettings,
|
||||
ProjectExecutionWorkspaceDefaultMode,
|
||||
ProjectExecutionWorkspacePolicy,
|
||||
SharedWorkspaceConcurrency,
|
||||
} from "@paperclipai/shared";
|
||||
import { asString, parseObject } from "../adapters/utils.js";
|
||||
|
||||
|
|
@ -110,6 +111,7 @@ export function parseProjectExecutionWorkspacePolicy(raw: unknown): ProjectExecu
|
|||
typeof parsed.defaultProjectWorkspaceId === "string" ? parsed.defaultProjectWorkspaceId : undefined;
|
||||
const allowIssueOverride =
|
||||
typeof parsed.allowIssueOverride === "boolean" ? parsed.allowIssueOverride : undefined;
|
||||
const sharedWorkspaceConcurrency = parseSharedWorkspaceConcurrency(parsed.sharedWorkspaceConcurrency);
|
||||
const normalizedDefaultMode = (() => {
|
||||
if (
|
||||
defaultMode === "shared_workspace" ||
|
||||
|
|
@ -125,6 +127,7 @@ export function parseProjectExecutionWorkspacePolicy(raw: unknown): ProjectExecu
|
|||
})();
|
||||
return {
|
||||
enabled,
|
||||
...(sharedWorkspaceConcurrency ? { sharedWorkspaceConcurrency } : {}),
|
||||
...(normalizedDefaultMode ? { defaultMode: normalizedDefaultMode } : {}),
|
||||
...(allowIssueOverride !== undefined ? { allowIssueOverride } : {}),
|
||||
...(defaultProjectWorkspaceId ? { defaultProjectWorkspaceId } : {}),
|
||||
|
|
@ -169,6 +172,7 @@ export function parseIssueExecutionWorkspaceSettings(
|
|||
const parsed = parseObject(raw);
|
||||
if (Object.keys(parsed).length === 0) return null;
|
||||
const workspaceStrategy = parseExecutionWorkspaceStrategy(parsed.workspaceStrategy);
|
||||
const sharedWorkspaceConcurrency = parseSharedWorkspaceConcurrency(parsed.sharedWorkspaceConcurrency);
|
||||
const mode = asString(parsed.mode, "");
|
||||
const normalizedMode = (() => {
|
||||
if (
|
||||
|
|
@ -200,6 +204,7 @@ export function parseIssueExecutionWorkspaceSettings(
|
|||
...(normalizedMode
|
||||
? { mode: normalizedMode as IssueExecutionWorkspaceSettings["mode"] }
|
||||
: {}),
|
||||
...(sharedWorkspaceConcurrency ? { sharedWorkspaceConcurrency } : {}),
|
||||
...(options.includeEnvironmentId && (typeof parsed.environmentId === "string" || parsed.environmentId === null)
|
||||
? { environmentId: parsed.environmentId }
|
||||
: {}),
|
||||
|
|
@ -309,6 +314,19 @@ export function resolveExecutionWorkspaceMode(input: {
|
|||
return "shared_workspace";
|
||||
}
|
||||
|
||||
function parseSharedWorkspaceConcurrency(raw: unknown): SharedWorkspaceConcurrency | undefined {
|
||||
return raw === "auto" || raw === "serialize" || raw === "allow" ? raw : undefined;
|
||||
}
|
||||
|
||||
export function resolveSharedWorkspaceConcurrency(input: {
|
||||
projectPolicy: ProjectExecutionWorkspacePolicy | null;
|
||||
issueSettings: IssueExecutionWorkspaceSettings | null;
|
||||
}): SharedWorkspaceConcurrency {
|
||||
return input.issueSettings?.sharedWorkspaceConcurrency
|
||||
?? (input.projectPolicy?.enabled ? input.projectPolicy.sharedWorkspaceConcurrency : undefined)
|
||||
?? "auto";
|
||||
}
|
||||
|
||||
export function buildExecutionWorkspaceAdapterConfig(input: {
|
||||
agentConfig: Record<string, unknown>;
|
||||
projectPolicy: ProjectExecutionWorkspacePolicy | null;
|
||||
|
|
|
|||
|
|
@ -193,6 +193,7 @@ import {
|
|||
resolveEffectiveWorkspaceStrategyType,
|
||||
resolveExecutionWorkspaceEnvironmentId,
|
||||
resolveExecutionWorkspaceMode,
|
||||
resolveSharedWorkspaceConcurrency,
|
||||
selectEnvironmentExecutionWorkspaceSettings,
|
||||
WORKSPACE_WORKTREE_REQUIRES_PROJECT_CODE,
|
||||
WORKSPACE_WORKTREE_REQUIRES_PROJECT_MESSAGE,
|
||||
|
|
@ -13457,8 +13458,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
.where(and(eq(issues.companyId, agent.companyId), eq(issues.id, issueContext.id), isNull(issues.responsibleUserId)));
|
||||
issueContext = { ...issueContext, responsibleUserId };
|
||||
}
|
||||
const parsedProjectExecutionWorkspacePolicy = parseProjectExecutionWorkspacePolicy(
|
||||
projectContext?.executionWorkspacePolicy,
|
||||
);
|
||||
const projectExecutionWorkspacePolicy = gateProjectExecutionWorkspacePolicy(
|
||||
parseProjectExecutionWorkspacePolicy(projectContext?.executionWorkspacePolicy),
|
||||
parsedProjectExecutionWorkspacePolicy,
|
||||
isolatedWorkspacesEnabled,
|
||||
);
|
||||
const trustPreset = resolveCoreTrustPreset({
|
||||
|
|
@ -13653,41 +13657,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
});
|
||||
const effectiveExecutionWorkspaceMode: ReturnType<typeof resolveExecutionWorkspaceMode> =
|
||||
requestedExecutionWorkspaceMode;
|
||||
// Serialize shared-workspace execution: two runs mutating the same project
|
||||
// working tree concurrently corrupt each other's uncommitted state, so a
|
||||
// run whose issue targets a busy shared workspace is deferred (rescheduled
|
||||
// retry) instead of dispatched, and keeps deferring until the workspace
|
||||
// frees — an adapter never dispatches alongside a live holder. Deadlock
|
||||
// safety comes from the holder query itself: a holder silent past
|
||||
// WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS stops counting (recovery's
|
||||
// silent-run escalation is already reaping it), so a zombie can only delay
|
||||
// work, never park it forever. This covers non-assignee runs (comment and
|
||||
// review wakes) too — their deferral records that the run never executed
|
||||
// under assignee-ship, so the retry promotion gate does not cancel it as a
|
||||
// reassignment.
|
||||
if (issueRef?.projectWorkspaceId && effectiveExecutionWorkspaceMode === "shared_workspace") {
|
||||
const workspaceHolder = await findSharedWorkspaceHolder({
|
||||
companyId: agent.companyId,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
excludeIssueId: issueRef.id,
|
||||
excludeRunId: run.id,
|
||||
honorIsolatedWorkspaceModes: isolatedWorkspacesEnabled,
|
||||
});
|
||||
if (workspaceHolder) {
|
||||
throw new WorkspaceBusyDeferral({
|
||||
holder: workspaceHolder,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
deferralAttempt:
|
||||
run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON
|
||||
? (run.scheduledRetryAttempt ?? 0)
|
||||
: 0,
|
||||
wasIssueAssignee: issueContext?.assigneeAgentId === agent.id,
|
||||
});
|
||||
}
|
||||
}
|
||||
const executionPolicy = { executionMode: (await instanceSettings.getGeneral()).executionMode };
|
||||
const executionPolicy = { executionMode: resolvedInstanceSettings.general.executionMode };
|
||||
const executionForcedToKubernetes = isExecutionForcedToKubernetes(executionPolicy);
|
||||
let selectedEnvironmentId = environmentResolution.environmentId;
|
||||
if (isExecutionForcedToKubernetes(executionPolicy)) {
|
||||
if (executionForcedToKubernetes) {
|
||||
let kubernetesEnvironment = await environmentsSvc.findKubernetesEnvironment(agent.companyId);
|
||||
if (!kubernetesEnvironment) {
|
||||
// Lazy recovery for companies created after the startup bootstrap ran
|
||||
|
|
@ -13751,6 +13724,79 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
}
|
||||
selectedEnvironmentId = kubernetesEnvironment.id;
|
||||
}
|
||||
const selectedEnvironmentForConfig = selectedEnvironmentId === localEnvironment.id
|
||||
? localEnvironment
|
||||
: selectedEnvironmentId
|
||||
? await environmentsSvc.getById(selectedEnvironmentId)
|
||||
: null;
|
||||
const sharedWorkspaceConcurrency = resolveSharedWorkspaceConcurrency({
|
||||
projectPolicy: projectExecutionWorkspacePolicy,
|
||||
issueSettings: issueExecutionWorkspaceSettings,
|
||||
});
|
||||
// A live holder is always consulted for shared workspaces. Depending on policy and the final
|
||||
// execution target it either remains the existing deferral gate or becomes dispatch context.
|
||||
// Holder staleness and the workspace_busy retry ladder are intentionally unchanged for every
|
||||
// path that serializes.
|
||||
if (issueRef?.projectWorkspaceId && effectiveExecutionWorkspaceMode === "shared_workspace") {
|
||||
const workspaceHolder = await findSharedWorkspaceHolder({
|
||||
companyId: agent.companyId,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
excludeIssueId: issueRef.id,
|
||||
excludeRunId: run.id,
|
||||
honorIsolatedWorkspaceModes: isolatedWorkspacesEnabled,
|
||||
});
|
||||
if (workspaceHolder) {
|
||||
const environmentDriver = selectedEnvironmentForConfig?.driver ?? null;
|
||||
const shouldSerialize = sharedWorkspaceConcurrency === "serialize"
|
||||
|| (
|
||||
sharedWorkspaceConcurrency === "auto"
|
||||
&& (
|
||||
executionForcedToKubernetes
|
||||
|| (environmentDriver !== "local" && environmentDriver !== "ssh")
|
||||
)
|
||||
);
|
||||
if (shouldSerialize) {
|
||||
throw new WorkspaceBusyDeferral({
|
||||
holder: workspaceHolder,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
deferralAttempt:
|
||||
run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON
|
||||
? (run.scheduledRetryAttempt ?? 0)
|
||||
: 0,
|
||||
wasIssueAssignee: issueContext?.assigneeAgentId === agent.id,
|
||||
});
|
||||
}
|
||||
|
||||
const holderIssueLabel = workspaceHolder.issueIdentifier ?? workspaceHolder.issueId;
|
||||
const concurrentWorkspaceNote =
|
||||
`shared workspace is concurrently held by run ${workspaceHolder.runId} (issue ${holderIssueLabel}); `
|
||||
+ "expect concurrent mutations, coordinate via commits";
|
||||
const appendConcurrentWorkspaceNote = (value: unknown) => {
|
||||
const existing = typeof value === "string" ? value.trimEnd() : "";
|
||||
return existing ? `${existing}\n${concurrentWorkspaceNote}` : concurrentWorkspaceNote;
|
||||
};
|
||||
context.paperclipTaskMarkdown = appendConcurrentWorkspaceNote(context.paperclipTaskMarkdown);
|
||||
if (typeof context.paperclipTaskMarkdownCompact === "string") {
|
||||
context.paperclipTaskMarkdownCompact = appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdownCompact,
|
||||
);
|
||||
}
|
||||
logger.info(
|
||||
{
|
||||
event: "shared_workspace_concurrent_dispatch",
|
||||
runId: run.id,
|
||||
issueId: issueRef.id,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
holderRunId: workspaceHolder.runId,
|
||||
holderIssueId: workspaceHolder.issueId,
|
||||
sharedWorkspaceConcurrency,
|
||||
environmentDriver,
|
||||
executionForcedToKubernetes,
|
||||
},
|
||||
"Dispatching alongside a live shared-workspace holder",
|
||||
);
|
||||
}
|
||||
}
|
||||
const workspaceManagedConfig = buildExecutionWorkspaceAdapterConfig({
|
||||
agentConfig: config,
|
||||
projectPolicy: projectExecutionWorkspacePolicy,
|
||||
|
|
@ -13796,11 +13842,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
});
|
||||
const configSnapshot = buildExecutionWorkspaceConfigSnapshot(mergedConfig, selectedEnvironmentId);
|
||||
const executionRunConfig = stripWorkspaceRuntimeFromExecutionRunConfig(mergedConfig);
|
||||
const selectedEnvironmentForConfig = selectedEnvironmentId === localEnvironment.id
|
||||
? localEnvironment
|
||||
: selectedEnvironmentId
|
||||
? await environmentsSvc.getById(selectedEnvironmentId)
|
||||
: null;
|
||||
const runScopedMentionedSkillKeys = await resolveRunScopedMentionedSkillKeys({
|
||||
db,
|
||||
companyId: agent.companyId,
|
||||
|
|
|
|||
Loading…
Reference in New Issue