feat: add ChatGPT OAuth agent boundary
Some checks failed
Publish Phase 3 workspace images / Publish agent image (push) Failing after 20s
Publish Phase 3 workspace images / Publish workspace-web image (push) Failing after 23s
Publish Phase 3 workspace images / Publish workspace-host image (push) Failing after 19s

This commit is contained in:
Varin Narula
2026-09-20 02:55:33 +07:00
parent 6cad93b332
commit e7b8980b98
22 changed files with 473 additions and 74 deletions

View File

@@ -32,6 +32,9 @@ jobs:
- name: workspace-web
dockerfile: Dockerfile.workspace
image: rokcet-workspace-web
- name: agent
dockerfile: Dockerfile.workspace
image: rokcet-agent
- name: workspace-host
dockerfile: Dockerfile.workspace-host
image: rokcet-workspace-host

View File

@@ -24,6 +24,7 @@ ARG BUN_VERSION=1.4.2
ENV DEBIAN_FRONTEND=noninteractive \
NODE_ENV=production \
HOME=/home/workspace \
CODEX_HOME=/home/workspace/.codex \
ROKCET_NODE_WORKSPACE_ROOT=/workspace/project \
PLAYWRIGHT_BROWSERS_PATH=/ms-playwright \
XDG_CACHE_HOME=/home/workspace/.cache \

View File

@@ -1,7 +1,7 @@
#!/bin/sh
set -eu
required_environment='ROKCET_NODE_URL ROKCET_NODE_REGISTRATION_CREDENTIAL ROKCET_NODE_WORKSPACE_ID ROKCET_NODE_CODING_SESSION_ID ROKCET_NODE_WORKSPACE_INSTANCE_ID ROKCET_NODE_ID ROKCET_NODE_INSTANCE_ID'
required_environment='ROKCET_NODE_URL ROKCET_NODE_REGISTRATION_CREDENTIAL ROKCET_NODE_ROLE ROKCET_NODE_WORKSPACE_ID ROKCET_NODE_CODING_SESSION_ID ROKCET_NODE_WORKSPACE_INSTANCE_ID ROKCET_NODE_ID ROKCET_NODE_INSTANCE_ID'
for environment_name in ${required_environment}; do
eval "environment_value=\${${environment_name}:-}"
@@ -23,4 +23,20 @@ if [ ! -d "${workspace_root}" ]; then
fi
umask 077
case "${ROKCET_NODE_ROLE}" in
agent)
codex_home="${CODEX_HOME:-/home/workspace/.codex}"
mkdir -p "${codex_home}"
if [ ! -f "${codex_home}/config.toml" ]; then
printf '%s\n' 'cli_auth_credentials_store = "file"' '[shell_environment_policy]' 'inherit = "core"' > "${codex_home}/config.toml"
chmod 0600 "${codex_home}/config.toml"
fi
;;
workspace) ;;
*)
echo "ROKCET_NODE_ROLE must be agent or workspace" >&2
exit 64
;;
esac
exec "$@"

View File

@@ -18,12 +18,15 @@ export function rokcetNodeOptionsFromEnvironment(): RokcetNodeClientOptions {
}
const workspaceRoot = path.resolve(requiredEnvironment('ROKCET_NODE_WORKSPACE_ROOT'));
if (workspaceRoot === path.parse(workspaceRoot).root) throw new Error('ROKCET_NODE_WORKSPACE_ROOT cannot be a filesystem root');
const role = process.env.ROKCET_NODE_ROLE?.trim() || 'agent';
if (role !== 'agent' && role !== 'workspace') throw new Error('ROKCET_NODE_ROLE must be agent or workspace');
return {
url: parsedUrl.toString(),
credential: requiredEnvironment('ROKCET_NODE_REGISTRATION_CREDENTIAL'),
workspaceId: requiredEnvironment('ROKCET_NODE_WORKSPACE_ID'),
codingSessionId: process.env.ROKCET_NODE_CODING_SESSION_ID?.trim() || undefined,
workspaceInstanceId: process.env.ROKCET_NODE_WORKSPACE_INSTANCE_ID?.trim() || undefined,
role,
workspaceRoot,
nodeId: requiredEnvironment('ROKCET_NODE_ID'),
instanceId: requiredEnvironment('ROKCET_NODE_INSTANCE_ID'),

View File

@@ -12,6 +12,7 @@ import tenantRouter from './routes/tenant';
import bucketsRouter from './routes/buckets';
import gitRouter from './routes/git';
import codingSessionsRouter from './routes/codingSessions';
import codexAccountRouter from './routes/codexAccount';
import { requireRokcetAuth } from './auth';
import { developerMcpClientMetadata } from './services/bucketsMcpOAuth';
import { attachRokcetNodeGateway } from './services/rokcetNodeConnection';
@@ -46,6 +47,7 @@ export function createApp() {
app.use('/api', requireRokcetAuth, bucketsRouter);
app.use('/api', requireRokcetAuth, gitRouter);
app.use('/api', requireRokcetAuth, codingSessionsRouter);
app.use('/api', requireRokcetAuth, codexAccountRouter);
// Preview compilation accepts generated source and therefore stays behind the
// Rokcet product session. Rendering is request-scoped and serverless-safe.

View File

@@ -0,0 +1,93 @@
import { Router } from 'express';
import {
getCodingSessionRecord,
getGitRepositoryRecord,
getProjectRecord,
type RokcetCodingSessionRecord,
type RokcetGitRepositoryRecord,
type RokcetProjectRecord,
} from '../services/rokcetDb';
import { rokcetAgentNodeRuntime } from '../services/rokcetNodeRuntime';
import { workspaceScheduler } from '../services/workspaceScheduler';
const router = Router();
type Context = { project: RokcetProjectRecord; repository: RokcetGitRepositoryRecord; session: RokcetCodingSessionRecord };
function context(req: any): Context {
const auth = req.rokcetAuth;
if (auth.isDemo) throw Object.assign(new Error('Codex login is unavailable in demo sessions'), { status: 403, code: 'DEMO_READ_ONLY' });
const project = getProjectRecord(auth.user.id, String(req.params.projectId || ''));
if (!project || !auth.workspaces.some((workspace: any) => workspace.id === project.workspaceId)) {
throw Object.assign(new Error('Project not found'), { status: 404, code: 'PROJECT_NOT_FOUND' });
}
const repository = getGitRepositoryRecord(project.id);
if (!repository || repository.status !== 'ready') throw Object.assign(new Error('Project repository is not ready'), { status: 409, code: 'REPOSITORY_NOT_READY' });
const session = getCodingSessionRecord(project.id, String(req.params.sessionId || ''));
if (!session) throw Object.assign(new Error('Coding session not found'), { status: 404, code: 'SESSION_NOT_FOUND' });
return { project, repository, session };
}
function scope(scoped: Context, leaseId: string) {
return {
userId: scoped.project.ownerUserId,
workspaceId: scoped.project.workspaceId,
projectId: scoped.project.id,
codingSessionId: scoped.session.id,
workspaceLeaseId: leaseId,
};
}
async function agent(scoped: Context, allocate: boolean) {
if (allocate) await workspaceScheduler.ensure(scoped);
const deadline = Date.now() + (allocate ? 30_000 : 0);
do {
const requester = rokcetAgentNodeRuntime.get(scoped.project.workspaceId, scoped.session.id);
if (requester?.healthy) return requester;
if (!allocate) break;
await new Promise((resolve) => setTimeout(resolve, 100));
} while (Date.now() < deadline);
throw Object.assign(new Error('The Codex agent is not connected'), { status: 503, code: 'CODEX_AGENT_UNAVAILABLE' });
}
function sendError(res: any, error: any): void {
res.status(error?.status || 500).json({
success: false,
code: error?.code || 'CODEX_ACCOUNT_ERROR',
error: error?.message || 'Codex account operation failed',
});
}
router.post('/projects/:projectId/sessions/:sessionId/codex/login', async (req, res) => {
try {
const scoped = context(req);
const requester = await agent(scoped, true);
const result = await requester.request(scope(scoped, requester.leaseId), 'agent.account.login.start', {});
if (result.type !== 'chatgptDeviceCode' || typeof result.loginId !== 'string' || typeof result.verificationUrl !== 'string' || typeof result.userCode !== 'string') {
throw Object.assign(new Error('Codex did not return a ChatGPT device authorization challenge'), { code: 'CODEX_LOGIN_CHALLENGE_INVALID' });
}
res.status(201).json({ success: true, login: result });
} catch (error) { sendError(res, error); }
});
router.get('/projects/:projectId/sessions/:sessionId/codex/account', async (req, res) => {
try {
const scoped = context(req);
const requester = await agent(scoped, false);
const result = await requester.request(scope(scoped, requester.leaseId), 'agent.account.read', {
refreshToken: req.query.refresh === 'true',
});
res.json({ success: true, ...result });
} catch (error) { sendError(res, error); }
});
router.delete('/projects/:projectId/sessions/:sessionId/codex/account', async (req, res) => {
try {
const scoped = context(req);
const requester = await agent(scoped, false);
await requester.request(scope(scoped, requester.leaseId), 'agent.account.logout', {});
res.status(204).end();
} catch (error) { sendError(res, error); }
});
export default router;

View File

@@ -1,6 +1,8 @@
import crypto from 'node:crypto';
import type { ServerNotification } from '../generated/codex/0.154.0-alpha.6.2/ts/ServerNotification';
import type { ServerRequest } from '../generated/codex/0.154.0-alpha.6.2/ts/ServerRequest';
import type { GetAccountResponse } from '../generated/codex/0.154.0-alpha.6.2/ts/v2/GetAccountResponse';
import type { LoginAccountResponse } from '../generated/codex/0.154.0-alpha.6.2/ts/v2/LoginAccountResponse';
import type {
CodingAgentAdapter,
CodingAgentAdapterSession,
@@ -98,6 +100,21 @@ export class CodexAppServerAdapter implements CodingAgentAdapter {
}
}
async readAccount(refreshToken = false): Promise<GetAccountResponse> {
await this.manager.start();
return this.manager.request('account/read', { refreshToken });
}
async startChatGptLogin(): Promise<LoginAccountResponse> {
await this.manager.start();
return this.manager.request('account/login/start', { type: 'chatgptDeviceCode' });
}
async logoutAccount(): Promise<void> {
await this.manager.start();
await this.manager.request('account/logout', {});
}
async startSession(input: CodingAgentStartSessionInput, onEvent: CodingAgentEventHandler): Promise<CodingAgentAdapterSession> {
await this.manager.start();
const response: any = await this.manager.request('thread/start', {

View File

@@ -1,4 +1,5 @@
import path from 'node:path';
import crypto from 'node:crypto';
import { execFile } from 'node:child_process';
import { chown, mkdir, rm, writeFile } from 'node:fs/promises';
import { promisify } from 'node:util';
@@ -38,25 +39,36 @@ function resources(spec: WorkspaceAllocationSpec): string[] {
];
}
function workspaceNodeResources(): string[] {
const memory = String(256 * 1024 * 1024);
return ['--cpus', '0.25', '--memory', memory, '--memory-swap', memory, '--pids-limit', '64'];
}
export class DockerWorkspaceRuntime {
private readonly network = process.env.ROKCET_WORKSPACE_DOCKER_NETWORK?.trim() || 'rokcet-workspaces';
async allocate(spec: WorkspaceAllocationSpec): Promise<{ containerId: string }> {
async allocate(spec: WorkspaceAllocationSpec): Promise<{ containerId: string; agentContainerId: string }> {
const instanceId = safeId(spec.instance.id, 'workspace instance id');
const containerName = `rokcet-ws-${instanceId}`;
const agentContainerName = `rokcet-agent-${instanceId}`;
const projectVolume = `${containerName}-project`;
const codexVolume = `${containerName}-codex`;
const cacheVolume = `${containerName}-cache`;
const codexVolume = `rokcet-codex-user-${crypto.createHash('sha256').update(spec.userId).digest('hex').slice(0, 32)}`;
const credentialRoot = path.join(stateRoot(), 'credentials', instanceId);
await this.ensureNetwork();
await Promise.all([projectVolume, codexVolume, cacheVolume].map((name) => docker(['volume', 'create', name])));
await Promise.all([projectVolume, cacheVolume, codexVolume].map((name) => docker(['volume', 'create', name])));
await docker([
'run', '--rm', '--user', '0:0', '--network', 'none',
'--mount', `type=volume,src=${projectVolume},dst=/workspace/project`,
'--mount', `type=volume,src=${codexVolume},dst=/home/workspace/.codex`,
'--mount', `type=volume,src=${cacheVolume},dst=/home/workspace/.cache`,
'--entrypoint', '/bin/sh', spec.instance.imageReference,
'-c', 'chown 10001:10001 /workspace/project /home/workspace/.codex /home/workspace/.cache',
'-c', 'chown 10001:10001 /workspace/project /home/workspace/.cache',
]);
await docker([
'run', '--rm', '--user', '0:0', '--network', 'none',
'--mount', `type=volume,src=${codexVolume},dst=/home/workspace/.codex`,
'--entrypoint', '/bin/sh', spec.agentImageReference,
'-c', 'chown 10001:10001 /home/workspace/.codex && chmod 0700 /home/workspace/.codex',
]);
let credentialMount: string[] = [];
@@ -74,14 +86,18 @@ export class DockerWorkspaceRuntime {
gitEnvironment = ['--env', 'GIT_SSH_COMMAND=ssh -i /run/rokcet/git/id_ed25519 -o IdentitiesOnly=yes -o UserKnownHostsFile=/run/rokcet/git/known_hosts -o StrictHostKeyChecking=yes'];
}
const commonMounts = [
const checkoutMounts = [
'--mount', `type=volume,src=${projectVolume},dst=/workspace/project`,
...credentialMount,
];
const commonMounts = [
'--mount', `type=volume,src=${projectVolume},dst=/workspace/project`,
'--mount', `type=volume,src=${cacheVolume},dst=/home/workspace/.cache`,
];
try {
await docker([
'run', '--rm', '--user', '10001:10001', '--network', this.network,
...commonMounts, ...gitEnvironment,
...checkoutMounts, ...gitEnvironment,
'--env', `ROKCET_GIT_CLONE_URL=${spec.source.cloneUrl}`,
'--env', `ROKCET_GIT_BRANCH=${spec.source.branchName}`,
'--env', `ROKCET_GIT_HEAD_SHA=${spec.source.headCommitSha}`,
@@ -90,46 +106,72 @@ export class DockerWorkspaceRuntime {
]);
await dockerIgnoreFailure(['rm', '--force', containerName]);
await dockerIgnoreFailure(['rm', '--force', agentContainerName]);
const containerId = await docker([
'create', '--name', containerName,
'--label', 'dev.rokcet.workspace=true',
'--label', `dev.rokcet.workspace.instance=${instanceId}`,
'--network', this.network,
'--read-only', '--cap-drop', 'ALL', '--security-opt', 'no-new-privileges',
...workspaceNodeResources(),
'--tmpfs', '/tmp:rw,nosuid,nodev,size=536870912,uid=10001,gid=10001',
...commonMounts,
'--env', `ROKCET_NODE_URL=${spec.nodes.workspace.url}`,
'--env', `ROKCET_NODE_REGISTRATION_CREDENTIAL=${spec.nodes.workspace.credential}`,
'--env', 'ROKCET_NODE_ROLE=workspace',
'--env', `ROKCET_NODE_WORKSPACE_ID=${spec.instance.workspaceId}`,
'--env', `ROKCET_NODE_CODING_SESSION_ID=${spec.instance.codingSessionId}`,
'--env', `ROKCET_NODE_WORKSPACE_INSTANCE_ID=${spec.instance.id}`,
'--env', `ROKCET_NODE_ID=${spec.nodes.workspace.nodeId}`,
'--env', `ROKCET_NODE_INSTANCE_ID=${spec.nodes.workspace.instanceId}`,
'--env', `ROKCET_NODE_BUILD=workspace-image:${spec.instance.imageReference}`,
spec.instance.imageReference,
]);
await docker(['start', containerId]);
const agentContainerId = await docker([
'create', '--name', agentContainerName,
'--label', 'dev.rokcet.agent=true',
'--label', `dev.rokcet.workspace.instance=${instanceId}`,
'--network', this.network,
'--read-only', '--cap-drop', 'ALL', '--security-opt', 'no-new-privileges',
...resources(spec),
'--tmpfs', '/tmp:rw,nosuid,nodev,size=536870912,uid=10001,gid=10001',
...commonMounts,
'--mount', `type=volume,src=${codexVolume},dst=/home/workspace/.codex`,
'--mount', `type=volume,src=${cacheVolume},dst=/home/workspace/.cache`,
...gitEnvironment,
'--env', `ROKCET_NODE_URL=${spec.node.url}`,
'--env', `ROKCET_NODE_REGISTRATION_CREDENTIAL=${spec.node.credential}`,
'--env', 'CODEX_HOME=/home/workspace/.codex',
'--env', `ROKCET_NODE_URL=${spec.nodes.agent.url}`,
'--env', `ROKCET_NODE_REGISTRATION_CREDENTIAL=${spec.nodes.agent.credential}`,
'--env', 'ROKCET_NODE_ROLE=agent',
'--env', `ROKCET_NODE_WORKSPACE_ID=${spec.instance.workspaceId}`,
'--env', `ROKCET_NODE_CODING_SESSION_ID=${spec.instance.codingSessionId}`,
'--env', `ROKCET_NODE_WORKSPACE_INSTANCE_ID=${spec.instance.id}`,
'--env', `ROKCET_NODE_ID=${spec.node.nodeId}`,
'--env', `ROKCET_NODE_INSTANCE_ID=${spec.node.instanceId}`,
'--env', `ROKCET_NODE_BUILD=workspace-image:${spec.instance.imageReference}`,
spec.instance.imageReference,
'--env', `ROKCET_NODE_ID=${spec.nodes.agent.nodeId}`,
'--env', `ROKCET_NODE_INSTANCE_ID=${spec.nodes.agent.instanceId}`,
'--env', `ROKCET_NODE_BUILD=agent-image:${spec.agentImageReference}`,
spec.agentImageReference,
]);
await docker(['start', containerId]);
return { containerId };
await docker(['start', agentContainerId]);
return { containerId, agentContainerId };
} catch (error) {
await dockerIgnoreFailure(['rm', '--force', containerName]);
await dockerIgnoreFailure(['rm', '--force', agentContainerName]);
throw error;
}
}
async stop(instance: { id: string; containerId: string | null }): Promise<void> {
async stop(instance: { id: string; containerId: string | null; agentContainerId?: string | null }): Promise<void> {
safeId(instance.id, 'workspace instance id');
if (instance.containerId) await dockerIgnoreFailure(['stop', '--time', '10', instance.containerId]);
if (instance.agentContainerId) await dockerIgnoreFailure(['stop', '--time', '10', instance.agentContainerId]);
}
async destroy(instance: { id: string; containerId: string | null }): Promise<void> {
async destroy(instance: { id: string; containerId: string | null; agentContainerId?: string | null }): Promise<void> {
const instanceId = safeId(instance.id, 'workspace instance id');
const containerName = `rokcet-ws-${instanceId}`;
const agentContainerName = `rokcet-agent-${instanceId}`;
await dockerIgnoreFailure(['rm', '--force', instance.containerId || containerName]);
await Promise.all(['project', 'codex', 'cache'].map((suffix) => dockerIgnoreFailure(['volume', 'rm', `${containerName}-${suffix}`])));
await dockerIgnoreFailure(['rm', '--force', instance.agentContainerId || agentContainerName]);
await Promise.all(['project', 'cache'].map((suffix) => dockerIgnoreFailure(['volume', 'rm', `${containerName}-${suffix}`])));
await rm(path.join(stateRoot(), 'credentials', instanceId), { recursive: true, force: true });
}

View File

@@ -261,6 +261,7 @@ export interface RokcetNodeLeaseRecord {
workspaceId: string;
codingSessionId: string | null;
workspaceInstanceId: string | null;
nodeRole: 'agent' | 'workspace';
registrationJti: string | null;
protocolVersion: string;
capabilities: unknown[];
@@ -315,6 +316,7 @@ export interface RokcetWorkspaceInstanceRecord {
imageReference: string;
status: string;
containerId: string | null;
agentContainerId: string | null;
gitCredentialId: string | null;
workspacePath: string;
cpuLimitMillis: number;
@@ -916,6 +918,27 @@ function applySchemaMigrations(conn: DatabaseSync): void {
throw error;
}
}
if (!applied.has(6)) {
db.exec('BEGIN IMMEDIATE');
try {
if (!hasColumn(db, 'rokcet_node_leases', 'node_role')) {
db.exec("ALTER TABLE rokcet_node_leases ADD COLUMN node_role TEXT NOT NULL DEFAULT 'agent';");
}
if (!hasColumn(db, 'rokcet_workspace_instances', 'agent_container_id')) {
db.exec('ALTER TABLE rokcet_workspace_instances ADD COLUMN agent_container_id TEXT;');
}
db.exec(`
CREATE INDEX IF NOT EXISTS ix_rokcet_node_leases_session_role_status
ON rokcet_node_leases(coding_session_id, node_role, status, expires_at);
`);
db.prepare('INSERT INTO rokcet_schema_migrations (version, name, applied_at) VALUES (6, ?, ?)')
.run('split_agent_workspace_nodes', new Date().toISOString());
db.exec('COMMIT');
} catch (error) {
db.exec('ROLLBACK');
throw error;
}
}
}
function rowToUser(row: any): RokcetUserRecord {
@@ -1786,6 +1809,7 @@ function rowToNodeLease(row: any): RokcetNodeLeaseRecord {
return {
id: row.id, nodeId: row.node_id, instanceId: row.instance_id, workspaceId: row.workspace_id,
codingSessionId: row.coding_session_id || null, workspaceInstanceId: row.workspace_instance_id || null,
nodeRole: row.node_role === 'workspace' ? 'workspace' : 'agent',
registrationJti: row.registration_jti || null, protocolVersion: row.protocol_version,
capabilities: parseJsonArray(row.capabilities_json), nodeBuild: row.node_build,
schemaDigest: row.schema_digest || null, status: row.status, expiresAt: row.expires_at,
@@ -1822,6 +1846,7 @@ function rowToWorkspaceInstance(row: any): RokcetWorkspaceInstanceRecord {
codingSessionId: row.coding_session_id, hostId: row.host_id || null,
generation: Number(row.generation), imageReference: row.image_reference,
status: row.status, containerId: row.container_id || null,
agentContainerId: row.agent_container_id || null,
gitCredentialId: row.git_credential_id || null,
workspacePath: row.workspace_path, cpuLimitMillis: Number(row.cpu_limit_millis),
memoryLimitBytes: Number(row.memory_limit_bytes), pidsLimit: Number(row.pids_limit),
@@ -2266,6 +2291,7 @@ export function activateNodeLease(input: {
workspaceId: string;
codingSessionId?: string | null;
workspaceInstanceId?: string | null;
nodeRole: 'agent' | 'workspace';
registrationJti: string;
registrationExpiresAt: string;
protocolVersion: string;
@@ -2288,21 +2314,22 @@ export function activateNodeLease(input: {
const id = existing?.id || crypto.randomUUID();
conn.prepare(`
INSERT INTO rokcet_node_leases (
id, node_id, instance_id, workspace_id, coding_session_id, workspace_instance_id,
id, node_id, instance_id, workspace_id, coding_session_id, workspace_instance_id, node_role,
registration_jti, protocol_version,
capabilities_json, node_build, schema_digest, status, expires_at,
last_heartbeat_at, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?, ?)
ON CONFLICT(node_id, instance_id, workspace_id) DO UPDATE SET
coding_session_id = excluded.coding_session_id,
workspace_instance_id = excluded.workspace_instance_id,
node_role = excluded.node_role,
registration_jti = excluded.registration_jti, protocol_version = excluded.protocol_version,
capabilities_json = excluded.capabilities_json, node_build = excluded.node_build,
schema_digest = excluded.schema_digest, status = 'active', expires_at = excluded.expires_at,
last_heartbeat_at = excluded.last_heartbeat_at, updated_at = excluded.updated_at
`).run(
id, input.nodeId, input.instanceId, input.workspaceId,
input.codingSessionId || null, input.workspaceInstanceId || null, input.registrationJti,
input.codingSessionId || null, input.workspaceInstanceId || null, input.nodeRole, input.registrationJti,
input.protocolVersion, JSON.stringify(input.capabilities), input.nodeBuild,
input.schemaDigest || null, input.expiresAt, now, now, now,
);
@@ -2319,6 +2346,16 @@ export function getNodeLease(leaseId: string): RokcetNodeLeaseRecord | null {
return row ? rowToNodeLease(row) : null;
}
export function listActiveNodeRolesForInstance(workspaceInstanceId: string): Array<'agent' | 'workspace'> {
const now = new Date().toISOString();
return (connection().prepare(`
SELECT DISTINCT node_role FROM rokcet_node_leases
WHERE workspace_instance_id = ? AND status = 'active' AND expires_at > ?
`).all(workspaceInstanceId, now) as Array<{ node_role: string }>).map((row) => (
row.node_role === 'workspace' ? 'workspace' : 'agent'
));
}
export function heartbeatNodeLease(leaseId: string, expiresAt: string): RokcetNodeLeaseRecord | null {
const now = new Date().toISOString();
connection().prepare(`
@@ -2500,6 +2537,7 @@ export function updateWorkspaceInstance(input: {
status: string;
hostId?: string | null;
containerId?: string | null;
agentContainerId?: string | null;
gitCredentialId?: string | null;
generation?: number;
errorCode?: string | null;
@@ -2512,7 +2550,7 @@ export function updateWorkspaceInstance(input: {
const now = new Date().toISOString();
connection().prepare(`
UPDATE rokcet_workspace_instances SET
status = ?, host_id = ?, container_id = ?, git_credential_id = ?, generation = ?,
status = ?, host_id = ?, container_id = ?, agent_container_id = ?, git_credential_id = ?, generation = ?,
error_code = ?, error_message = ?, last_activity_at = ?,
ready_at = ?, stopped_at = ?, updated_at = ?
WHERE id = ?
@@ -2520,6 +2558,7 @@ export function updateWorkspaceInstance(input: {
input.status,
input.hostId === undefined ? existing.hostId : input.hostId,
input.containerId === undefined ? existing.containerId : input.containerId,
input.agentContainerId === undefined ? existing.agentContainerId : input.agentContainerId,
input.gitCredentialId === undefined ? existing.gitCredentialId : input.gitCredentialId,
input.generation === undefined ? existing.generation : input.generation,
input.errorCode === undefined ? existing.errorCode : input.errorCode,

View File

@@ -43,6 +43,7 @@ export interface RokcetNodeClientOptions {
workspaceId: string;
codingSessionId?: string;
workspaceInstanceId?: string;
role?: 'agent' | 'workspace';
workspaceRoot: string;
nodeId: string;
instanceId: string;
@@ -56,7 +57,8 @@ type CachedResponse = RokcetNodeResponse['payload'];
export class RokcetNodeClient {
private socket: WebSocket | null = null;
private readonly adapter: CodingAgentAdapter;
private readonly role: 'agent' | 'workspace';
private readonly adapter: CodingAgentAdapter | null;
private welcome: RokcetNodeWelcome | null = null;
private heartbeatTimer: NodeJS.Timeout | null = null;
private nextEventSequence = 1;
@@ -66,7 +68,8 @@ export class RokcetNodeClient {
private closing = false;
constructor(private readonly options: RokcetNodeClientOptions) {
this.adapter = options.adapter || new CodexAppServerAdapter();
this.role = options.role || 'agent';
this.adapter = this.role === 'agent' ? (options.adapter || new CodexAppServerAdapter()) : null;
}
async connect(): Promise<RokcetNodeWelcome> {
@@ -95,11 +98,12 @@ export class RokcetNodeClient {
nodeBuild: this.options.nodeBuild || `rokcet-node/codex-${CODEX_APP_SERVER_VERSION}`,
platform: process.platform, architecture: process.arch,
protocolMin: ROKCET_NODE_PROTOCOL_VERSION, protocolMax: ROKCET_NODE_PROTOCOL_VERSION,
capabilities: [
{ name: 'agent' as const, version: 1, limits: { schemaDigest: CODEX_APP_SERVER_SCHEMA_DIGEST } },
{ name: 'git' as const, version: 1, limits: { maxFilesPerCommit: 250, maxCommitBytes: 10 * 1024 * 1024 } },
{ name: 'workspace.status' as const, version: 1 },
],
capabilities: this.role === 'agent'
? [{ name: 'agent' as const, version: 1, limits: { schemaDigest: CODEX_APP_SERVER_SCHEMA_DIGEST } }]
: [
{ name: 'git' as const, version: 1, limits: { maxFilesPerCommit: 250, maxCommitBytes: 10 * 1024 * 1024 } },
{ name: 'workspace.status' as const, version: 1 },
],
credential: this.options.credential,
},
};
@@ -127,7 +131,7 @@ export class RokcetNodeClient {
const socket = this.socket;
this.socket = null;
if (socket && socket.readyState < WebSocket.CLOSING) socket.close(1000, 'Rokcet Node client shutting down');
await this.adapter.shutdown();
await this.adapter?.shutdown();
}
private fail(error: Error): void {
@@ -139,7 +143,7 @@ export class RokcetNodeClient {
if (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING) {
socket.close(1011, 'Rokcet Node client failure');
}
void this.adapter.shutdown().finally(() => this.options.onDisconnect?.(error));
void (this.adapter?.shutdown() || Promise.resolve()).finally(() => this.options.onDisconnect?.(error));
}
private send(frame: RokcetNodeEnvelope): void {
@@ -159,7 +163,8 @@ export class RokcetNodeClient {
private handleWelcome(frame: RokcetNodeWelcome): void {
if (this.welcome) throw new Error('Control plane sent more than one welcome frame');
if (!frame.payload.workspaceLeaseId || !frame.payload.selectedCapabilities.some((item) => item.name === 'agent' && item.version === 1)) {
const requiredCapability = this.role === 'agent' ? 'agent' : 'workspace.status';
if (!frame.payload.workspaceLeaseId || !frame.payload.selectedCapabilities.some((item) => item.name === requiredCapability && item.version === 1)) {
throw new Error('Control plane selected an invalid Rokcet Node capability set');
}
this.welcome = frame;
@@ -235,6 +240,22 @@ export class RokcetNodeClient {
private async execute(frame: RokcetNodeRequest): Promise<Record<string, unknown>> {
const args = frame.payload.arguments;
if (this.role === 'workspace' && frame.payload.operation.startsWith('agent.')) {
throw Object.assign(new Error('Agent operation is unavailable on the workspace node'), { code: 'UNSUPPORTED_OPERATION' });
}
if (this.role === 'agent' && !frame.payload.operation.startsWith('agent.')) {
throw Object.assign(new Error('Workspace operation is unavailable on the agent node'), { code: 'UNSUPPORTED_OPERATION' });
}
if (this.role === 'agent' && frame.payload.operation === 'agent.account.read') {
return { ...await (this.adapter as CodexAppServerAdapter).readAccount(Boolean(args.refreshToken)) };
}
if (this.role === 'agent' && frame.payload.operation === 'agent.account.login.start') {
return { ...await (this.adapter as CodexAppServerAdapter).startChatGptLogin() };
}
if (this.role === 'agent' && frame.payload.operation === 'agent.account.logout') {
await (this.adapter as CodexAppServerAdapter).logoutAccount();
return {};
}
if (frame.payload.operation === 'git.status') {
return workspaceGitStatus(this.workspacePath(frame.scope, args.workspacePath));
}
@@ -248,6 +269,11 @@ export class RokcetNodeClient {
return workspaceStatus(this.workspacePath(frame.scope, args.workspacePath));
}
if (frame.payload.operation === 'agent.session.start') {
if (!this.adapter) throw Object.assign(new Error('Agent capability is unavailable on the workspace node'), { code: 'UNSUPPORTED_OPERATION' });
if (this.adapter instanceof CodexAppServerAdapter) {
const account = await this.adapter.readAccount(true);
if (!account.account) throw Object.assign(new Error('Connect a ChatGPT account before starting a coding session'), { code: 'CODEX_AUTH_REQUIRED' });
}
const input = args as unknown as CodingAgentStartSessionInput;
const session = await this.adapter.startSession({
...input, codingSessionId: frame.scope.codingSessionId,
@@ -256,6 +282,7 @@ export class RokcetNodeClient {
return { ...session };
}
if (frame.payload.operation === 'agent.session.resume') {
if (!this.adapter) throw Object.assign(new Error('Agent capability is unavailable on the workspace node'), { code: 'UNSUPPORTED_OPERATION' });
const input = args as unknown as CodingAgentStartSessionInput & { providerThreadId: string };
const session = await this.adapter.resumeSession(input.providerThreadId, {
...input, codingSessionId: frame.scope.codingSessionId,
@@ -264,6 +291,7 @@ export class RokcetNodeClient {
return { ...session };
}
if (frame.payload.operation === 'agent.turn.start') {
if (!this.adapter) throw Object.assign(new Error('Agent capability is unavailable on the workspace node'), { code: 'UNSUPPORTED_OPERATION' });
const input = args as unknown as CodingAgentStartTurnInput & { providerThreadId: string };
return { ...await this.adapter.startTurn(input.providerThreadId, {
...input, codingSessionId: frame.scope.codingSessionId,
@@ -271,13 +299,16 @@ export class RokcetNodeClient {
}) };
}
if (frame.payload.operation === 'agent.turn.interrupt') {
if (!this.adapter) throw Object.assign(new Error('Agent capability is unavailable on the workspace node'), { code: 'UNSUPPORTED_OPERATION' });
await this.adapter.interruptTurn(String(args.providerThreadId || ''), String(args.providerTurnId || ''));
return {};
}
if (frame.payload.operation === 'agent.review.start') {
if (!this.adapter) throw Object.assign(new Error('Agent capability is unavailable on the workspace node'), { code: 'UNSUPPORTED_OPERATION' });
return { ...await this.adapter.startReview(String(args.providerThreadId || ''), args.target as any, String(args.turnId || '')) };
}
if (frame.payload.operation === 'agent.approval.resolve') {
if (!this.adapter) throw Object.assign(new Error('Agent capability is unavailable on the workspace node'), { code: 'UNSUPPORTED_OPERATION' });
await this.adapter.resolveApproval(String(args.providerRequestId || ''), args.decision as any, (args.details || {}) as Record<string, unknown>);
return {};
}

View File

@@ -24,14 +24,18 @@ import {
completeNodeRequest,
createNodeRequest,
getWorkspaceInstance,
listActiveNodeRolesForInstance,
heartbeatNodeLease,
updateWorkspaceInstance,
type RokcetNodeLeaseRecord,
} from './rokcetDb';
import { verifyRokcetNodeRegistrationCredential } from './rokcetNodeCredentials';
import { rokcetNodeRuntime } from './rokcetNodeRuntime';
import { rokcetAgentNodeRuntime, rokcetNodeRuntime } from './rokcetNodeRuntime';
const REQUIRED_CAPABILITIES: RokcetNodeCapability[] = [{ name: 'agent', version: 1 }];
const REQUIRED_CAPABILITIES: Record<'agent' | 'workspace', RokcetNodeCapability[]> = {
agent: [{ name: 'agent', version: 1 }],
workspace: [{ name: 'git', version: 1 }, { name: 'workspace.status', version: 1 }],
};
const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000;
const DEFAULT_LEASE_TTL_MS = 45_000;
const DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
@@ -82,6 +86,7 @@ export class RokcetNodeConnection implements RokcetNodeRequester {
readonly workspaceId: string;
readonly nodeBuild: string;
readonly schemaDigest: string;
readonly role: 'agent' | 'workspace';
private pending = new Map<string, PendingRequest>();
private pendingByIdempotency = new Map<string, Promise<Record<string, unknown>>>();
private nextEventSequence = 1;
@@ -102,6 +107,7 @@ export class RokcetNodeConnection implements RokcetNodeRequester {
this.workspaceId = lease.workspaceId;
this.nodeBuild = lease.nodeBuild;
this.schemaDigest = lease.schemaDigest || '';
this.role = lease.nodeRole;
this.expiryTimer = setInterval(() => this.checkExpiry(), Math.min(leaseTtlMs, 5_000));
this.expiryTimer.unref();
socket.on('message', (data) => {
@@ -116,17 +122,25 @@ export class RokcetNodeConnection implements RokcetNodeRequester {
}
attachAdapter(): RemoteCodingAgentAdapter {
if (this.role !== 'agent') throw new Error('Workspace node cannot attach a coding-agent adapter');
if (this.adapter) return this.adapter;
this.adapter = new RemoteCodingAgentAdapter(this);
this.unregisterAdapter = this.lease.codingSessionId
? codingAgentRuntime.registerSession(this.lease.codingSessionId, this.adapter)
: codingAgentRuntime.register(this.workspaceId, this.adapter);
this.unregisterRuntime = rokcetNodeRuntime.register({
this.unregisterRuntime = rokcetAgentNodeRuntime.register({
workspaceId: this.workspaceId, codingSessionId: this.lease.codingSessionId, requester: this,
});
return this.adapter;
}
attachWorkspaceRuntime(): void {
if (this.role !== 'workspace' || this.unregisterRuntime) return;
this.unregisterRuntime = rokcetNodeRuntime.register({
workspaceId: this.workspaceId, codingSessionId: this.lease.codingSessionId, requester: this,
});
}
async request(
scope: RokcetNodeScope,
operation: string,
@@ -219,6 +233,7 @@ export class RokcetNodeConnection implements RokcetNodeRequester {
}
private async handleEvent(frame: RokcetNodeEvent): Promise<void> {
if (this.role !== 'agent') throw new Error('Workspace node cannot emit coding-agent events');
validateScope(frame.scope, this.lease);
if (frame.connectionSequence < this.nextEventSequence) {
this.send({ ...envelopeBase(), kind: 'ack', payload: { connectionSequence: frame.connectionSequence } });
@@ -304,11 +319,12 @@ export class RokcetNodeGateway {
}
get(workspaceId: string): RokcetNodeConnection | null {
return this.connections.get(`workspace:${workspaceId}`) || null;
return this.connections.get(`workspace:${workspaceId}:agent`)
|| this.connections.get(`workspace:${workspaceId}:workspace`) || null;
}
getForSession(codingSessionId: string): RokcetNodeConnection | null {
return this.connections.get(`session:${codingSessionId}`) || null;
getForSession(codingSessionId: string, role: 'agent' | 'workspace' = 'agent'): RokcetNodeConnection | null {
return this.connections.get(`session:${codingSessionId}:${role}`) || null;
}
async close(): Promise<void> {
@@ -339,29 +355,32 @@ export class RokcetNodeGateway {
|| !['allocating', 'starting', 'recovering'].includes(instance.status)
) throw new Error('Rokcet Node workspace instance scope mismatch');
}
const selectedCapabilities = negotiateRokcetNodeCapabilities(frame.payload.capabilities, REQUIRED_CAPABILITIES);
const agentCapability = selectedCapabilities.find((capability) => capability.name === 'agent')!;
const schemaDigest = typeof agentCapability.limits?.schemaDigest === 'string' ? agentCapability.limits.schemaDigest : '';
if (schemaDigest !== CODEX_APP_SERVER_SCHEMA_DIGEST) throw new Error('Rokcet Node agent schema digest mismatch');
const selectedCapabilities = negotiateRokcetNodeCapabilities(frame.payload.capabilities, REQUIRED_CAPABILITIES[claims.nodeRole]);
const agentCapability = selectedCapabilities.find((capability) => capability.name === 'agent');
const schemaDigest = typeof agentCapability?.limits?.schemaDigest === 'string' ? agentCapability.limits.schemaDigest : '';
if (claims.nodeRole === 'agent' && schemaDigest !== CODEX_APP_SERVER_SCHEMA_DIGEST) throw new Error('Rokcet Node agent schema digest mismatch');
const expiresAt = new Date(Date.now() + this.leaseTtlMs).toISOString();
const lease = activateNodeLease({
nodeId: frame.payload.nodeId, instanceId: frame.payload.instanceId,
workspaceId: claims.workspaceId, registrationJti: claims.jti,
codingSessionId: claims.codingSessionId || null,
workspaceInstanceId: claims.workspaceInstanceId || null,
nodeRole: claims.nodeRole,
registrationExpiresAt: new Date(claims.exp * 1000).toISOString(),
protocolVersion: ROKCET_NODE_PROTOCOL_VERSION, capabilities: selectedCapabilities,
nodeBuild: frame.payload.nodeBuild, schemaDigest, expiresAt,
});
const connectionKey = claims.codingSessionId
? `session:${claims.codingSessionId}` : `workspace:${claims.workspaceId}`;
? `session:${claims.codingSessionId}:${claims.nodeRole}` : `workspace:${claims.workspaceId}:${claims.nodeRole}`;
const previous = this.connections.get(connectionKey);
previous?.close(4002, 'A newer Rokcet Node connection replaced this connection');
const connection = new RokcetNodeConnection(socket, lease, this.requestTimeoutMs, this.leaseTtlMs);
this.connections.set(connectionKey, connection);
connection.attachAdapter();
if (claims.nodeRole === 'agent') connection.attachAdapter();
else connection.attachWorkspaceRuntime();
if (claims.workspaceInstanceId) {
updateWorkspaceInstance({
const roles = new Set(listActiveNodeRolesForInstance(claims.workspaceInstanceId));
if (roles.has('agent') && roles.has('workspace')) updateWorkspaceInstance({
instanceId: claims.workspaceInstanceId, status: 'ready', readyAt: new Date().toISOString(),
});
}

View File

@@ -9,6 +9,7 @@ export interface RokcetNodeRegistrationClaims {
workspaceId: string;
codingSessionId?: string;
workspaceInstanceId?: string;
nodeRole: 'agent' | 'workspace';
nodeId: string;
instanceId: string;
jti: string;
@@ -32,6 +33,7 @@ export function issueRokcetNodeRegistrationCredential(input: {
workspaceId: string;
codingSessionId?: string;
workspaceInstanceId?: string;
nodeRole?: 'agent' | 'workspace';
nodeId: string;
instanceId: string;
ttlSeconds?: number;
@@ -44,6 +46,7 @@ export function issueRokcetNodeRegistrationCredential(input: {
workspaceId: input.workspaceId,
...(input.codingSessionId ? { codingSessionId: input.codingSessionId } : {}),
...(input.workspaceInstanceId ? { workspaceInstanceId: input.workspaceInstanceId } : {}),
nodeRole: input.nodeRole || 'agent',
nodeId: input.nodeId,
instanceId: input.instanceId,
jti: crypto.randomUUID(),
@@ -80,6 +83,7 @@ export function verifyRokcetNodeRegistrationCredential(credential: string, now =
|| typeof claims.workspaceId !== 'string' || !claims.workspaceId
|| (claims.codingSessionId !== undefined && (typeof claims.codingSessionId !== 'string' || !claims.codingSessionId))
|| (claims.workspaceInstanceId !== undefined && (typeof claims.workspaceInstanceId !== 'string' || !claims.workspaceInstanceId))
|| !['agent', 'workspace'].includes(String(claims.nodeRole))
|| typeof claims.nodeId !== 'string' || !claims.nodeId
|| typeof claims.instanceId !== 'string' || !claims.instanceId
|| typeof claims.jti !== 'string' || !claims.jti

View File

@@ -19,3 +19,4 @@ class RokcetNodeRuntimeRegistry {
}
export const rokcetNodeRuntime = new RokcetNodeRuntimeRegistry();
export const rokcetAgentNodeRuntime = new RokcetNodeRuntimeRegistry();

View File

@@ -42,9 +42,10 @@ function validateHello(frame: WorkspaceHostHello): void {
for (const field of ['hostId', 'instanceId', 'displayName', 'region', 'hostBuild', 'nonce', 'signature'] as const) {
if (typeof payload[field] !== 'string' || !payload[field].trim()) throw new Error(`Workspace host ${field} is required`);
}
for (const field of ['cpuMillis', 'memoryBytes', 'diskBytes', 'maxInstances'] as const) {
for (const field of ['cpuMillis', 'memoryBytes', 'diskBytes'] as const) {
if (!Number.isSafeInteger(payload.capacity?.[field]) || payload.capacity[field] <= 0) throw new Error(`Workspace host ${field} is invalid`);
}
if (!Number.isSafeInteger(payload.capacity?.maxInstances) || payload.capacity.maxInstances < 0) throw new Error('Workspace host maxInstances is invalid');
}
type Pending = { resolve: (value: Record<string, unknown>) => void; reject: (error: Error) => void; timer: NodeJS.Timeout };
@@ -63,20 +64,22 @@ class WorkspaceHostConnection implements WorkspaceHostDriver {
socket.once('error', (error) => this.fail(error));
}
allocate(spec: WorkspaceAllocationSpec): Promise<{ containerId: string }> {
allocate(spec: WorkspaceAllocationSpec): Promise<{ containerId: string; agentContainerId: string }> {
return this.request('workspace.allocate', spec as unknown as Record<string, unknown>)
.then((result) => {
if (typeof result.containerId !== 'string' || !result.containerId) throw new Error('Workspace host returned no container id');
return { containerId: result.containerId };
if (typeof result.containerId !== 'string' || !result.containerId || typeof result.agentContainerId !== 'string' || !result.agentContainerId) {
throw new Error('Workspace host returned incomplete container ids');
}
return { containerId: result.containerId, agentContainerId: result.agentContainerId };
});
}
async stop(instance: RokcetWorkspaceInstanceRecord): Promise<void> {
await this.request('workspace.stop', { instanceId: instance.id, containerId: instance.containerId });
await this.request('workspace.stop', { instanceId: instance.id, containerId: instance.containerId, agentContainerId: instance.agentContainerId });
}
async destroy(instance: RokcetWorkspaceInstanceRecord): Promise<void> {
await this.request('workspace.destroy', { instanceId: instance.id, containerId: instance.containerId });
await this.request('workspace.destroy', { instanceId: instance.id, containerId: instance.containerId, agentContainerId: instance.agentContainerId });
}
close(reason = 'Workspace host connection replaced'): void {

View File

@@ -6,6 +6,7 @@ import type {
} from './rokcetDb';
import {
createWorkspaceInstance,
getWorkspaceInstance,
getWorkspaceInstanceForSession,
listWorkspaceHosts,
listWorkspaceInstancesForHost,
@@ -36,7 +37,9 @@ function requiredPublicNodeUrl(): string {
}
export interface WorkspaceAllocationSpec {
userId: string;
instance: RokcetWorkspaceInstanceRecord;
agentImageReference: string;
source: {
repositoryPath: string;
cloneUrl: string;
@@ -48,17 +51,25 @@ export interface WorkspaceAllocationSpec {
knownHosts: string | null;
};
};
node: {
url: string;
credential: string;
nodeId: string;
instanceId: string;
nodes: {
agent: {
url: string;
credential: string;
nodeId: string;
instanceId: string;
};
workspace: {
url: string;
credential: string;
nodeId: string;
instanceId: string;
};
};
}
export interface WorkspaceHostDriver {
readonly hostId: string;
allocate(spec: WorkspaceAllocationSpec): Promise<{ containerId: string }>;
allocate(spec: WorkspaceAllocationSpec): Promise<{ containerId: string; agentContainerId: string }>;
stop(instance: RokcetWorkspaceInstanceRecord): Promise<void>;
destroy(instance: RokcetWorkspaceInstanceRecord): Promise<void>;
}
@@ -131,11 +142,16 @@ export class WorkspaceScheduler {
errorCode: null, errorMessage: null,
readyAt: null, stoppedAt: null,
})!;
const nodeId = `workspace-${allocating.id}`;
const nodeInstanceId = `${allocating.id}-g${generation}`;
const workspaceNodeId = `workspace-${allocating.id}`;
const agentNodeId = `agent-${allocating.id}`;
const workspaceNodeInstanceId = `${allocating.id}-workspace-g${generation}`;
const agentNodeInstanceId = `${allocating.id}-agent-g${generation}`;
const nodeUrl = requiredPublicNodeUrl();
try {
const allocation = await driver.allocate({
userId: input.project.ownerUserId,
instance: allocating,
agentImageReference: process.env.ROKCET_AGENT_IMAGE?.trim() || 'rokcet-agent:dev',
source: {
repositoryPath: input.repository.repositoryPath,
cloneUrl: gitCredential.cloneUrl,
@@ -146,17 +162,29 @@ export class WorkspaceScheduler {
knownHosts: gitCredential.knownHosts,
},
},
node: {
url: requiredPublicNodeUrl(), nodeId, instanceId: nodeInstanceId,
credential: issueRokcetNodeRegistrationCredential({
workspaceId: input.project.workspaceId, codingSessionId: input.session.id,
workspaceInstanceId: allocating.id, nodeId, instanceId: nodeInstanceId,
ttlSeconds: 300,
}),
nodes: {
workspace: {
url: nodeUrl, nodeId: workspaceNodeId, instanceId: workspaceNodeInstanceId,
credential: issueRokcetNodeRegistrationCredential({
workspaceId: input.project.workspaceId, codingSessionId: input.session.id,
workspaceInstanceId: allocating.id, nodeRole: 'workspace',
nodeId: workspaceNodeId, instanceId: workspaceNodeInstanceId, ttlSeconds: 300,
}),
},
agent: {
url: nodeUrl, nodeId: agentNodeId, instanceId: agentNodeInstanceId,
credential: issueRokcetNodeRegistrationCredential({
workspaceId: input.project.workspaceId, codingSessionId: input.session.id,
workspaceInstanceId: allocating.id, nodeRole: 'agent',
nodeId: agentNodeId, instanceId: agentNodeInstanceId, ttlSeconds: 300,
}),
},
},
});
const current = getWorkspaceInstance(allocating.id);
return updateWorkspaceInstance({
instanceId: allocating.id, status: 'starting', containerId: allocation.containerId,
instanceId: allocating.id, status: current?.status === 'ready' ? 'ready' : 'starting', containerId: allocation.containerId,
agentContainerId: allocation.agentContainerId,
})!;
} catch (error: any) {
if (gitCredential.credentialId) {
@@ -194,6 +222,7 @@ export class WorkspaceScheduler {
updateWorkspaceInstance({
instanceId: instance.id, status: cleaned ? 'destroyed' : 'cleanup_pending',
containerId: cleaned ? null : instance.containerId, gitCredentialId: null,
agentContainerId: cleaned ? null : instance.agentContainerId,
stoppedAt: new Date().toISOString(),
...(cleaned ? { hostId: null, errorCode: null, errorMessage: null } : {
errorCode: 'WORKSPACE_CLEANUP_PENDING',

View File

@@ -41,6 +41,7 @@ import { normalizeProjectContract } from '../shared/projectContracts';
import { ROKCET_SESSION_EXPIRED_EVENT, rokcetFetch, setRokcetSessionExpiryNotificationsEnabled, setRokcetSessionKind } from './services/rokcetFetch';
import { useCodingAgentSession } from './hooks/useCodingAgentSession';
import { ACTIVE_CODING_AGENT_TURN_STATUSES } from '../shared/codingAgentEvents';
import { startChatGptDeviceLogin } from './services/codingAgent';
function loadDemoStorage<T>(key: string, fallback: T): T {
try {
@@ -1536,6 +1537,24 @@ export default function App() {
});
return;
} catch (error: any) {
if (error?.code === 'CODEX_AUTH_REQUIRED') {
try {
const login = await startChatGptDeviceLogin(runProject.id, currentSess.id);
const loginMessage: ChatMessage = {
id: `msg-agent-chatgpt-auth-${Date.now()}`, sessionId: currentSess.id, role: 'assistant',
content: `Connect your ChatGPT account to continue. Open [ChatGPT device authorization](${login.verificationUrl}) and enter code **${login.userCode}**. Then run your request again.`,
timestamp: new Date().toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }),
};
const loginSession = { ...currentSess, updatedAt: new Date().toISOString(), messages: [...messagesBeforeWorking, loginMessage] };
setSessions((current) => current.map((session) => session.id === currentSess!.id ? loginSession : session));
setActiveSession(loginSession);
window.open(login.verificationUrl, '_blank', 'noopener,noreferrer');
setIsExecuting(false);
return;
} catch (loginError: any) {
error = loginError;
}
}
const errorMessage: ChatMessage = {
id: `msg-agent-error-${Date.now()}`, sessionId: currentSess.id, role: 'system',
content: `⚠️ **Coding agent could not start**: ${error?.message || 'Unknown error'}`,

View File

@@ -17,6 +17,13 @@ export type CodingAgentStartResponse =
type: 'connect_buckets'; workspaceId: string; projectId: string; projectName: string; capabilities: string[];
} };
export interface ChatGptDeviceLogin {
type: 'chatgptDeviceCode';
loginId: string;
verificationUrl: string;
userCode: string;
}
function sessionPath(projectId: string, codingSessionId: string): string {
return `/api/projects/${encodeURIComponent(projectId)}/sessions/${encodeURIComponent(codingSessionId)}`;
}
@@ -89,6 +96,21 @@ export async function acceptCodingAgentChanges(input: {
return readApi(response);
}
export async function startChatGptDeviceLogin(projectId: string, codingSessionId: string): Promise<ChatGptDeviceLogin> {
const response = await rokcetFetch(`${sessionPath(projectId, codingSessionId)}/codex/login`, { method: 'POST' });
return (await readApi<{ login: ChatGptDeviceLogin }>(response)).login;
}
export async function readChatGptAccount(projectId: string, codingSessionId: string): Promise<Record<string, unknown> | null> {
const response = await rokcetFetch(`${sessionPath(projectId, codingSessionId)}/codex/account`);
return (await readApi<{ account: Record<string, unknown> | null }>(response)).account;
}
export async function disconnectChatGptAccount(projectId: string, codingSessionId: string): Promise<void> {
const response = await rokcetFetch(`${sessionPath(projectId, codingSessionId)}/codex/account`, { method: 'DELETE' });
if (!response.ok) await readApi(response);
}
export function subscribeCodingAgentEvents(input: {
projectId: string;
codingSessionId: string;

View File

@@ -19,6 +19,23 @@ test('removes control-plane and secret-bearing environment values from the Codex
assert.deepEqual(environment, { PATH: '/usr/bin', HOME: '/workspace-home', LANG: 'C.UTF-8' });
});
test('starts ChatGPT device authorization and exposes only account metadata', async () => {
const adapter = new CodexAppServerAdapter({
command: process.execPath, args: [fixture], skipVersionCheck: true, requestTimeoutMs: 5_000,
});
assert.equal((await adapter.readAccount()).account, null);
assert.deepEqual(await adapter.startChatGptLogin(), {
type: 'chatgptDeviceCode', loginId: 'login-1',
verificationUrl: 'https://auth.openai.com/device', userCode: 'ABCD-EFGH',
});
assert.deepEqual((await adapter.readAccount()).account, {
type: 'chatgpt', email: 'user@example.com', planType: 'plus',
});
await adapter.logoutAccount();
assert.equal((await adapter.readAccount()).account, null);
await adapter.shutdown();
});
test('normalizes pinned App Server lifecycle and approval events', async () => {
const adapter = new CodexAppServerAdapter({
command: process.execPath,

View File

@@ -2,6 +2,7 @@ import readline from 'node:readline';
const input = readline.createInterface({ input: process.stdin, crlfDelay: Infinity });
const send = (value) => process.stdout.write(`${JSON.stringify(value)}\n`);
let account = null;
input.on('line', (line) => {
const message = JSON.parse(line);
@@ -12,6 +13,15 @@ input.on('line', (line) => {
}
return;
}
if (message.method === 'account/read') return send({ id: message.id, result: { account, requiresOpenaiAuth: true } });
if (message.method === 'account/login/start') {
account = { type: 'chatgpt', email: 'user@example.com', planType: 'plus' };
return send({ id: message.id, result: { type: 'chatgptDeviceCode', loginId: 'login-1', verificationUrl: 'https://auth.openai.com/device', userCode: 'ABCD-EFGH' } });
}
if (message.method === 'account/logout') {
account = null;
return send({ id: message.id, result: {} });
}
if (message.method === 'thread/start') {
return send({ id: message.id, result: { thread: { id: 'thread-1', sessionId: 'thread-1' } } });
}

View File

@@ -54,7 +54,7 @@ test('extracts project metadata and removes legacy source history', () => {
const migrated = new DatabaseSync(databasePath);
const versions = migrated.prepare('SELECT version FROM rokcet_schema_migrations ORDER BY version').all() as Array<{ version: number }>;
assert.deepEqual(versions.map((row) => Number(row.version)), [1, 2, 3, 4, 5]);
assert.deepEqual(versions.map((row) => Number(row.version)), [1, 2, 3, 4, 5, 6]);
const agentTables = migrated.prepare(`
SELECT name FROM sqlite_master WHERE type = 'table' AND name LIKE 'rokcet_agent_%' ORDER BY name
`).all() as Array<{ name: string }>;

View File

@@ -14,6 +14,7 @@ import type {
} from '../shared/codingAgentEvents';
import { ROKCET_NODE_PROTOCOL_VERSION } from '../shared/rokcetNodeProtocol';
import { codingAgentRuntime } from '../server/services/codingAgentRuntime';
import { rokcetNodeRuntime } from '../server/services/rokcetNodeRuntime';
import { CODEX_APP_SERVER_SCHEMA_DIGEST } from '../server/services/codexAppServerAdapter';
import { attachRokcetNodeGateway, type RokcetNodeGateway } from '../server/services/rokcetNodeConnection';
import { RokcetNodeClient } from '../server/services/rokcetNodeClient';
@@ -99,6 +100,7 @@ after(async () => {
await gateway.close();
await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
codingAgentRuntime.clearForTests();
rokcetNodeRuntime.clearForTests();
resetRokcetDbForTests();
rmSync(root, { recursive: true, force: true });
});
@@ -121,6 +123,25 @@ test('registration credentials can be restricted to one session instance', () =>
assert.equal(claims.workspaceInstanceId, 'workspace-instance-1');
});
test('keeps workspace nodes credential-free and rejects agent operations', async () => {
const workspaceCredential = issueRokcetNodeRegistrationCredential({
workspaceId, codingSessionId, nodeRole: 'workspace',
nodeId: 'workspace-node', instanceId: 'workspace-node-1',
});
const workspaceClient = new RokcetNodeClient({
url, credential: workspaceCredential, workspaceId, codingSessionId,
workspaceRoot, nodeId: 'workspace-node', instanceId: 'workspace-node-1', role: 'workspace',
});
const welcome = await workspaceClient.connect();
const requester = rokcetNodeRuntime.get(workspaceId, codingSessionId);
assert.ok(requester);
await assert.rejects(() => requester.request({
userId, workspaceId, projectId, codingSessionId,
workspaceLeaseId: welcome.payload.workspaceLeaseId,
}, 'agent.account.read', {}), /Agent operation is unavailable/);
await workspaceClient.close();
});
test('negotiates a remote adapter and transports ordered agent events', async () => {
const adapter = codingAgentRuntime.get(workspaceId);
assert.ok(adapter);

View File

@@ -30,6 +30,13 @@ function positiveInteger(name: string, fallback?: number): number {
return Number(value);
}
function nonNegativeInteger(name: string, fallback = 0): number {
const raw = process.env[name]?.trim();
const value = raw ? Number(raw) : fallback;
if (!Number.isSafeInteger(value) || Number(value) < 0) throw new Error(`${name} must be a non-negative integer`);
return Number(value);
}
function envelopeBase() {
return { protocolVersion: WORKSPACE_HOST_PROTOCOL_VERSION, messageId: crypto.randomUUID(), sentAt: new Date().toISOString() } as const;
}
@@ -72,7 +79,7 @@ export function workspaceHostOptionsFromEnvironment(): WorkspaceHostOptions {
cpuMillis: positiveInteger('ROKCET_WORKSPACE_HOST_CPU_MILLIS', 3_000),
memoryBytes: positiveInteger('ROKCET_WORKSPACE_HOST_MEMORY_BYTES', 3 * 1024 * 1024 * 1024),
diskBytes: positiveInteger('ROKCET_WORKSPACE_HOST_DISK_BYTES', 40 * 1024 * 1024 * 1024),
maxInstances: positiveInteger('ROKCET_WORKSPACE_HOST_MAX_INSTANCES', 1),
maxInstances: nonNegativeInteger('ROKCET_WORKSPACE_HOST_MAX_INSTANCES', 1),
};
}
@@ -180,10 +187,10 @@ export class WorkspaceHostClient {
if (frame.payload.operation === 'workspace.allocate') {
payload = { ok: true, result: await this.runtime.allocate(args as WorkspaceAllocationSpec) };
} else if (frame.payload.operation === 'workspace.stop') {
await this.runtime.stop({ id: String(args.instanceId || ''), containerId: typeof args.containerId === 'string' ? args.containerId : null });
await this.runtime.stop({ id: String(args.instanceId || ''), containerId: typeof args.containerId === 'string' ? args.containerId : null, agentContainerId: typeof args.agentContainerId === 'string' ? args.agentContainerId : null });
payload = { ok: true, result: {} };
} else if (frame.payload.operation === 'workspace.destroy') {
await this.runtime.destroy({ id: String(args.instanceId || ''), containerId: typeof args.containerId === 'string' ? args.containerId : null });
await this.runtime.destroy({ id: String(args.instanceId || ''), containerId: typeof args.containerId === 'string' ? args.containerId : null, agentContainerId: typeof args.agentContainerId === 'string' ? args.agentContainerId : null });
payload = { ok: true, result: {} };
} else {
throw Object.assign(new Error('Unsupported workspace host operation'), { code: 'UNSUPPORTED_OPERATION' });