mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 20:46:28 +01:00
fix(workspaces): persist readiness state and validate ports (#9408)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Execution workspaces can run managed services that must report reliable lifecycle and readiness state > - Service startup previously waited for readiness before committing the starting row, making concurrent control actions see stale state > - Fixed service ports also needed clearer configuration and ownership diagnostics to avoid cross-workspace collisions > - This pull request persists startup state before readiness, validates port ownership, and exposes configurable service ports in the workspace UI > - The benefit is dependable service controls and actionable diagnostics when workspace runtimes start slowly or compete for ports ## Linked Issues or Issue Description ### What happened? Slow-starting workspace services could remain invisible to concurrent stop/restart controls until readiness completed, and fixed-port conflicts lacked enough ownership context for safe repair. ### Expected behavior A starting service is persisted immediately, control operations can observe it, configured ports are editable, and conflicts identify the owning process/workspace. ### Steps to reproduce 1. Configure a workspace service that delays binding its HTTP port. 2. Start the service and immediately request another control action. 3. Observe stale persisted state before this change. 4. Configure two workspaces for the same fixed port and observe limited conflict diagnostics. ### Paperclip version or commit `origin/master` at `02e2dd271` ### Deployment mode Local dev; built from source; not adapter-specific; database-backed workspace runtime state. ## What Changed - Commit the `starting` runtime-service row before waiting for readiness and transition it after the probe completes. - Add port-owner inspection and cross-workspace conflict details to local service supervision. - Preserve configurable runtime service ports through workspace configuration updates. - Surface service-port editing and validation in the execution workspace details UI. - Add server and UI regression coverage for slow readiness, concurrent controls, port persistence, and conflict diagnostics. ## Verification - `vitest --project @paperclipai/server src/__tests__/workspace-runtime.test.ts src/__tests__/execution-workspaces-service.test.ts` — 118 tests passed. - `vitest --project @paperclipai/ui src/pages/ExecutionWorkspaceDetail.service-ports.test.ts` — 4 tests passed. - `node scripts/check-token-gates.mjs` — all token gates clean. ## Risks - Moderate risk: changes touch workspace service lifecycle persistence and local process/port inspection. - No schema migration is required; tests exercise slow readiness, concurrent control, persisted ports, and cross-workspace conflicts. > 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 Codex, GPT-5.3 Codex, reasoning with repository tool use and code execution; context-window size was not exposed by the runtime. ## 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:
DottaandPaperclip authored and GitHub committed
1 parent
a739a8dfba
commit
e4e12bfb89
7 files changed
+836
-79
No files matched your search
@@ -24,6 +24,8 @@ import {
|
||||
createLocalServiceKey,
|
||||
findLocalServiceRegistryRecordByRuntimeServiceId,
|
||||
findAdoptableLocalService,
|
||||
isLocalServiceProcessInWorkspace,
|
||||
readLocalServiceProcessCwd,
|
||||
readLocalServicePortOwner,
|
||||
removeLocalServiceRegistryRecord,
|
||||
terminateLocalService,
|
||||
@@ -127,6 +129,11 @@ interface RuntimeServiceRecord extends RuntimeServiceRef {
|
||||
processGroupId: number | null;
|
||||
}
|
||||
|
||||
type LocalRuntimeServiceStart = {
|
||||
record: RuntimeServiceRecord;
|
||||
readiness: Promise<void>;
|
||||
};
|
||||
|
||||
type StoppedRuntimeServiceReuseCandidate = {
|
||||
id: string;
|
||||
port: number | null;
|
||||
@@ -3785,7 +3792,7 @@ export function normalizeAdapterManagedRuntimeServices(input: {
|
||||
});
|
||||
}
|
||||
|
||||
async function startLocalRuntimeService(input: {
|
||||
type StartLocalRuntimeServiceInput = {
|
||||
db?: Db;
|
||||
runId: string;
|
||||
leaseRunId?: string | null;
|
||||
@@ -3800,7 +3807,9 @@ async function startLocalRuntimeService(input: {
|
||||
reuseKey: string | null;
|
||||
scopeType: "project_workspace" | "execution_workspace" | "run" | "agent";
|
||||
scopeId: string | null;
|
||||
}): Promise<RuntimeServiceRecord> {
|
||||
};
|
||||
|
||||
async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): Promise<LocalRuntimeServiceStart> {
|
||||
const leaseRunId = input.leaseRunId === undefined ? input.runId : input.leaseRunId;
|
||||
const startedByRunId = input.startedByRunId === undefined ? input.runId : input.startedByRunId;
|
||||
const identity = resolveRuntimeServiceReuseIdentity({
|
||||
@@ -3899,48 +3908,61 @@ async function startLocalRuntimeService(input: {
|
||||
await removeLocalServiceRegistryRecord(adoptedRecord.serviceKey);
|
||||
} else {
|
||||
return {
|
||||
id: adoptedRecord.runtimeServiceId ?? randomUUID(),
|
||||
companyId: input.agent.companyId,
|
||||
projectId: input.workspace.projectId,
|
||||
projectWorkspaceId: input.workspace.workspaceId,
|
||||
executionWorkspaceId: input.executionWorkspaceId ?? null,
|
||||
issueId: input.issue?.id ?? null,
|
||||
serviceName,
|
||||
status: "running",
|
||||
lifecycle,
|
||||
scopeType: input.scopeType,
|
||||
scopeId: input.scopeId,
|
||||
reuseKey: input.reuseKey,
|
||||
command,
|
||||
cwd: serviceCwd,
|
||||
port: adoptedRecord.port ?? port,
|
||||
url: adoptedRecord.url ?? url,
|
||||
provider: "local_process",
|
||||
providerRef: String(adoptedRecord.pid),
|
||||
ownerAgentId: input.agent.id ?? null,
|
||||
startedByRunId,
|
||||
lastUsedAt: new Date().toISOString(),
|
||||
startedAt: adoptedRecord.startedAt,
|
||||
stoppedAt: null,
|
||||
stopPolicy,
|
||||
healthStatus: "healthy",
|
||||
reused: true,
|
||||
db: input.db,
|
||||
child: null,
|
||||
leaseRunIds: leaseRunId ? new Set([leaseRunId]) : new Set(),
|
||||
idleTimer: null,
|
||||
envFingerprint,
|
||||
serviceKey,
|
||||
profileKind: "workspace-runtime",
|
||||
processGroupId: adoptedRecord.processGroupId ?? null,
|
||||
record: {
|
||||
id: adoptedRecord.runtimeServiceId ?? randomUUID(),
|
||||
companyId: input.agent.companyId,
|
||||
projectId: input.workspace.projectId,
|
||||
projectWorkspaceId: input.workspace.workspaceId,
|
||||
executionWorkspaceId: input.executionWorkspaceId ?? null,
|
||||
issueId: input.issue?.id ?? null,
|
||||
serviceName,
|
||||
status: "running",
|
||||
lifecycle,
|
||||
scopeType: input.scopeType,
|
||||
scopeId: input.scopeId,
|
||||
reuseKey: input.reuseKey,
|
||||
command,
|
||||
cwd: serviceCwd,
|
||||
port: adoptedRecord.port ?? port,
|
||||
url: adoptedRecord.url ?? url,
|
||||
provider: "local_process",
|
||||
providerRef: String(adoptedRecord.pid),
|
||||
ownerAgentId: input.agent.id ?? null,
|
||||
startedByRunId,
|
||||
lastUsedAt: new Date().toISOString(),
|
||||
startedAt: adoptedRecord.startedAt,
|
||||
stoppedAt: null,
|
||||
stopPolicy,
|
||||
healthStatus: "healthy",
|
||||
reused: true,
|
||||
db: input.db,
|
||||
child: null,
|
||||
leaseRunIds: leaseRunId ? new Set([leaseRunId]) : new Set(),
|
||||
idleTimer: null,
|
||||
envFingerprint,
|
||||
serviceKey,
|
||||
profileKind: "workspace-runtime",
|
||||
processGroupId: adoptedRecord.processGroupId ?? null,
|
||||
},
|
||||
readiness: Promise.resolve(),
|
||||
};
|
||||
}
|
||||
}
|
||||
if (identityPort) {
|
||||
const ownerPid = await readLocalServicePortOwner(identityPort);
|
||||
const ownerPid = await readLocalServicePortOwner(identityPort);
|
||||
if (ownerPid) {
|
||||
const ownerCwd = await readLocalServiceProcessCwd(ownerPid);
|
||||
const ownerIsInWorkspace = ownerCwd
|
||||
? await isLocalServiceProcessInWorkspace(ownerCwd, serviceCwd)
|
||||
: null;
|
||||
const ownerDescription = ownerCwd ? `pid ${ownerPid} (cwd: ${ownerCwd})` : `pid ${ownerPid} (cwd unavailable)`;
|
||||
if (ownerIsInWorkspace === false) {
|
||||
throw new Error(
|
||||
`Runtime service "${serviceName}" could not start because port ${identityPort} has a cross-workspace port conflict with ${ownerDescription}; requested workspace: ${serviceCwd}. Stop the other service or configure a different port.`,
|
||||
);
|
||||
}
|
||||
throw new Error(
|
||||
`Runtime service "${serviceName}" could not start because port ${identityPort} is already in use by pid ${ownerPid}`,
|
||||
`Runtime service "${serviceName}" could not start because port ${identityPort} is already in use by ${ownerDescription}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -3974,18 +3996,7 @@ async function startLocalRuntimeService(input: {
|
||||
if (input.onLog) await input.onLog("stderr", `[service:${serviceName}] ${text}`);
|
||||
});
|
||||
|
||||
try {
|
||||
await Promise.race([
|
||||
waitForReadiness({ service: input.service, serviceName, command, url }),
|
||||
spawnErrorPromise,
|
||||
]);
|
||||
} catch (err) {
|
||||
terminateChildProcess(child);
|
||||
throw new Error(
|
||||
`Failed to start runtime service "${serviceName}": ${err instanceof Error ? err.message : String(err)}${stderrExcerpt ? ` | stderr: ${stderrExcerpt.trim()}` : ""}`,
|
||||
);
|
||||
}
|
||||
|
||||
const nowIso = new Date().toISOString();
|
||||
const record: RuntimeServiceRecord = {
|
||||
id: stoppedReuseCandidate?.id ?? randomUUID(),
|
||||
companyId: input.agent.companyId,
|
||||
@@ -3994,7 +4005,7 @@ async function startLocalRuntimeService(input: {
|
||||
executionWorkspaceId: input.executionWorkspaceId ?? null,
|
||||
issueId: input.issue?.id ?? null,
|
||||
serviceName,
|
||||
status: "running",
|
||||
status: "starting",
|
||||
lifecycle,
|
||||
scopeType: input.scopeType,
|
||||
scopeId: input.scopeId,
|
||||
@@ -4007,11 +4018,11 @@ async function startLocalRuntimeService(input: {
|
||||
providerRef: child.pid ? String(child.pid) : null,
|
||||
ownerAgentId: input.agent.id ?? null,
|
||||
startedByRunId,
|
||||
lastUsedAt: new Date().toISOString(),
|
||||
startedAt: new Date().toISOString(),
|
||||
lastUsedAt: nowIso,
|
||||
startedAt: nowIso,
|
||||
stoppedAt: null,
|
||||
stopPolicy,
|
||||
healthStatus: "healthy",
|
||||
healthStatus: "unknown",
|
||||
reused: false,
|
||||
db: input.db,
|
||||
child,
|
||||
@@ -4052,7 +4063,37 @@ async function startLocalRuntimeService(input: {
|
||||
});
|
||||
}
|
||||
|
||||
return record;
|
||||
const readinessPromise = Promise.race([
|
||||
waitForReadiness({ service: input.service, serviceName, command, url }),
|
||||
spawnErrorPromise,
|
||||
]).then(async () => {
|
||||
record.status = "running";
|
||||
record.healthStatus = "healthy";
|
||||
record.lastUsedAt = new Date().toISOString();
|
||||
record.stoppedAt = null;
|
||||
await touchLocalServiceRegistryRecord(record.serviceKey, {
|
||||
runtimeServiceId: record.id,
|
||||
lastSeenAt: record.lastUsedAt,
|
||||
});
|
||||
}).catch(async (err) => {
|
||||
terminateChildProcess(child);
|
||||
record.status = "stopped";
|
||||
record.healthStatus = "unhealthy";
|
||||
record.lastUsedAt = new Date().toISOString();
|
||||
record.stoppedAt = new Date().toISOString();
|
||||
await removeLocalServiceRegistryRecord(record.serviceKey).catch(() => undefined);
|
||||
throw new Error(
|
||||
`Failed to start runtime service "${serviceName}": ${err instanceof Error ? err.message : String(err)}${stderrExcerpt ? ` | stderr: ${stderrExcerpt.trim()}` : ""}`,
|
||||
);
|
||||
});
|
||||
|
||||
return { record, readiness: readinessPromise };
|
||||
}
|
||||
|
||||
async function startLocalRuntimeService(input: StartLocalRuntimeServiceInput): Promise<RuntimeServiceRecord> {
|
||||
const started = await spawnLocalRuntimeService(input);
|
||||
await started.readiness;
|
||||
return started.record;
|
||||
}
|
||||
|
||||
function scheduleIdleStop(record: RuntimeServiceRecord) {
|
||||
@@ -4332,14 +4373,23 @@ type StartRuntimeServicesForWorkspaceControlInput = {
|
||||
respectDesiredStates?: boolean;
|
||||
};
|
||||
|
||||
type WorkspaceControlStartBatch = {
|
||||
refs: RuntimeServiceRef[];
|
||||
pendingReadiness: LocalRuntimeServiceStart[];
|
||||
startedServiceIds: string[];
|
||||
};
|
||||
|
||||
async function startRuntimeServicesForWorkspaceControlUnlocked(
|
||||
input: StartRuntimeServicesForWorkspaceControlInput,
|
||||
rawServices: Record<string, unknown>[],
|
||||
invocationId: string,
|
||||
persistenceDb = input.db,
|
||||
registryDb = input.db,
|
||||
): Promise<RuntimeServiceRef[]> {
|
||||
options?: { deferReadiness?: boolean },
|
||||
): Promise<WorkspaceControlStartBatch> {
|
||||
const refs: RuntimeServiceRef[] = [];
|
||||
const pendingReadiness: LocalRuntimeServiceStart[] = [];
|
||||
const startedServiceIds: string[] = [];
|
||||
|
||||
for (const service of rawServices) {
|
||||
const { scopeType, scopeId } = resolveServiceScopeId({
|
||||
@@ -4377,9 +4427,7 @@ async function startRuntimeServicesForWorkspaceControlUnlocked(
|
||||
}
|
||||
}
|
||||
|
||||
// Manually controlled services are not tied to a heartbeat run lifecycle, so they do not
|
||||
// retain a run lease and never persist a startedByRunId foreign key.
|
||||
const record = await startLocalRuntimeService({
|
||||
const startInput: StartLocalRuntimeServiceInput = {
|
||||
db: persistenceDb,
|
||||
runId: invocationId,
|
||||
leaseRunId: null,
|
||||
@@ -4394,13 +4442,30 @@ async function startRuntimeServicesForWorkspaceControlUnlocked(
|
||||
reuseKey,
|
||||
scopeType,
|
||||
scopeId,
|
||||
});
|
||||
registerRuntimeService(registryDb, record);
|
||||
await persistRuntimeServiceRecord(persistenceDb, record);
|
||||
refs.push(toRuntimeServiceRef(record));
|
||||
};
|
||||
|
||||
// Manually controlled services are not tied to a heartbeat run lifecycle, so they do not
|
||||
// retain a run lease and never persist a startedByRunId foreign key.
|
||||
const started = options?.deferReadiness
|
||||
? await spawnLocalRuntimeService(startInput)
|
||||
: {
|
||||
record: await startLocalRuntimeService(startInput),
|
||||
readiness: Promise.resolve(),
|
||||
};
|
||||
registerRuntimeService(registryDb, started.record);
|
||||
await persistRuntimeServiceRecord(persistenceDb, started.record);
|
||||
refs.push(toRuntimeServiceRef(started.record));
|
||||
|
||||
if (options?.deferReadiness && !started.record.reused) {
|
||||
// Attach a rejection handler immediately; the caller awaits the same promise after
|
||||
// the DB transaction commits, but transaction failures may skip that wait path.
|
||||
started.readiness.catch(() => undefined);
|
||||
pendingReadiness.push(started);
|
||||
startedServiceIds.push(started.record.id);
|
||||
}
|
||||
}
|
||||
|
||||
return refs;
|
||||
return { refs, pendingReadiness, startedServiceIds };
|
||||
}
|
||||
|
||||
export async function startRuntimeServicesForWorkspaceControl(
|
||||
@@ -4416,12 +4481,17 @@ export async function startRuntimeServicesForWorkspaceControl(
|
||||
const invocationId = input.invocationId ?? randomUUID();
|
||||
|
||||
if (rawServices.length === 0 || !input.db || (!input.executionWorkspaceId && !input.workspace.workspaceId)) {
|
||||
return startRuntimeServicesForWorkspaceControlUnlocked(input, rawServices, invocationId);
|
||||
const batch = await startRuntimeServicesForWorkspaceControlUnlocked(input, rawServices, invocationId);
|
||||
return batch.refs;
|
||||
}
|
||||
|
||||
let startedRefs: RuntimeServiceRef[] = [];
|
||||
let startBatch: WorkspaceControlStartBatch = {
|
||||
refs: [],
|
||||
pendingReadiness: [],
|
||||
startedServiceIds: [],
|
||||
};
|
||||
try {
|
||||
return await input.db.transaction(async (tx) => {
|
||||
await input.db.transaction(async (tx) => {
|
||||
const txDb = tx as unknown as Db;
|
||||
|
||||
if (input.executionWorkspaceId) {
|
||||
@@ -4453,20 +4523,35 @@ export async function startRuntimeServicesForWorkspaceControl(
|
||||
}
|
||||
|
||||
// Branch reconciliation takes these same parent row locks before mutating
|
||||
// a recorded branch. Holding them until the running service row is
|
||||
// persisted closes the process-start window before the FK insert.
|
||||
startedRefs = await startRuntimeServicesForWorkspaceControlUnlocked(
|
||||
// a recorded branch. Persisting a `starting` service row before commit closes
|
||||
// the process-start window without holding the DB transaction for readiness.
|
||||
startBatch = await startRuntimeServicesForWorkspaceControlUnlocked(
|
||||
{ ...input, db: txDb },
|
||||
rawServices,
|
||||
invocationId,
|
||||
txDb,
|
||||
input.db,
|
||||
{ deferReadiness: true },
|
||||
);
|
||||
return startedRefs;
|
||||
});
|
||||
|
||||
for (const pending of startBatch.pendingReadiness) {
|
||||
try {
|
||||
await pending.readiness;
|
||||
await persistRuntimeServiceRecord(input.db, pending.record);
|
||||
} catch (error) {
|
||||
await persistRuntimeServiceRecord(input.db, pending.record).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
return startBatch.refs.map((ref) => {
|
||||
const record = runtimeServicesById.get(ref.id);
|
||||
return record ? toRuntimeServiceRef(record, { reused: ref.reused }) : ref;
|
||||
});
|
||||
} catch (error) {
|
||||
for (const ref of startedRefs) {
|
||||
await stopRuntimeService(ref.id).catch(() => undefined);
|
||||
for (const serviceId of startBatch.startedServiceIds) {
|
||||
await stopRuntimeService(serviceId).catch(() => undefined);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user