Skip to content

fix: scope OpenCode resume sessions by workspace #1007

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
CharlieZhao95 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: scope OpenCode resume sessions by workspace #1007

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions packages/api/src/domains/cats/services/agents/invocation/invoke-single-cat.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,6 +376,17 @@ function isUserVisibleSessionOutput(msg: AgentMessage): boolean {
return msg.type === 'text' || msg.type === 'tool_use' || msg.type === 'tool_result';
}

function normalizeSessionWorkspace(path: string | undefined): string | undefined {
const trimmed = path?.trim();
return trimmed ? resolve(trimmed) : undefined;
}

function sessionWorkspaceMatches(record: SessionRecord, workingDirectory: string | undefined): boolean {
const expected = normalizeSessionWorkspace(workingDirectory);
const actual = normalizeSessionWorkspace(record.workingDirectory);
return Boolean(expected && actual && expected === actual);
}

async function syncAntigravityRuntimeMetadata(input: {
runtimeSessionStore: IRuntimeSessionStore;
sessionChainStore: ISessionChainStore;
Expand Down Expand Up @@ -1071,6 +1082,32 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP
}
const workingProjectRoot = workingDirectory ? findMonorepoRoot(workingDirectory) : undefined;

if (provider === 'opencode' && sessionId && deps.sessionChainStore && sessionChainActive && !isBgCarrier) {
try {
const activeRec = await preflightRace(
Promise.resolve(deps.sessionChainStore.getActive(catId, threadId)),
'getActive:workspaceResumeGuard',
signal,
);
if (!activeRec || activeRec.cliSessionId !== sessionId || !sessionWorkspaceMatches(activeRec, workingDirectory)) {
log.debug(
{
catId,
threadId,
invocationId,
cliSessionId: sessionId,
recordWorkingDirectory: activeRec?.workingDirectory ?? null,
workingDirectory: workingDirectory ?? null,
},
'OpenCode resume session discarded because stored workspace does not match thread workspace',
);
sessionId = undefined;
}
} catch {
sessionId = undefined;
}
}

// Shared-state preflight — covers ALL cats (Claude/Codex/Gemini), vendor-agnostic.
// Three-layer defense model (shared-rules §14):
// L1 .githooks/pre-commit = hard block (prevents committing on wrong branch)
Expand Down Expand Up @@ -2171,6 +2208,7 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP
if (bgRec.cliSessionId !== msg.sessionId) {
await deps.sessionChainStore.update(bgRec.id, {
cliSessionId: msg.sessionId,
...(workingDirectory ? { workingDirectory } : {}),
...(params.continuityCapsule ? { continuityCapsule: params.continuityCapsule } : {}),
updatedAt: Date.now(),
});
Expand All @@ -2182,6 +2220,7 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP
} else {
const newRec = await deps.sessionChainStore.create({
cliSessionId: msg.sessionId,
...(workingDirectory ? { workingDirectory } : {}),
threadId,
catId,
userId,
Expand All @@ -2206,6 +2245,7 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP
// This is normal — NOT a "session replaced" event. Just update the tracked ID.
await deps.sessionChainStore.update(existing.id, {
cliSessionId: msg.sessionId,
...(workingDirectory ? { workingDirectory } : {}),
...(params.continuityCapsule ? { continuityCapsule: params.continuityCapsule } : {}),
updatedAt: Date.now(),
});
Expand Down Expand Up @@ -2280,6 +2320,7 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP
const inheritedFailures = existing.consecutiveRestoreFailures ?? 0;
const newRec = await deps.sessionChainStore.create({
cliSessionId: msg.sessionId,
...(workingDirectory ? { workingDirectory } : {}),
threadId,
catId,
userId,
Expand All @@ -2298,13 +2339,17 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP
}
} else if (params.continuityCapsule) {
await deps.sessionChainStore.update(existing.id, {
...(workingDirectory ? { workingDirectory } : {}),
continuityCapsule: params.continuityCapsule,
});
} else if (workingDirectory && existing.workingDirectory !== workingDirectory) {
await deps.sessionChainStore.update(existing.id, { workingDirectory });
}
} else {
// No active session (first invocation or previous was sealed)
const newRec = await deps.sessionChainStore.create({
cliSessionId: msg.sessionId,
...(workingDirectory ? { workingDirectory } : {}),
threadId,
catId,
userId,
Expand Down
4 changes: 4 additions & 0 deletions packages/api/src/domains/cats/services/stores/ports/SessionChainStore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import type { CatId, SessionRecord } from '@cat-cafe/shared';

export interface CreateSessionInput {
cliSessionId: string;
workingDirectory?: string;
threadId: string;
catId: CatId;
userId: string;
Expand All @@ -28,6 +29,7 @@ export type SessionRecordPatch = Partial<
Pick<
SessionRecord,
| 'cliSessionId'
| 'workingDirectory'
| 'status'
| 'contextHealth'
| 'lastUsage'
Expand Down Expand Up @@ -114,6 +116,7 @@ export class SessionChainStore implements ISessionChainStore {
const record: SessionRecord = {
id,
cliSessionId: input.cliSessionId,
...(input.workingDirectory ? { workingDirectory: input.workingDirectory } : {}),
threadId: input.threadId,
catId: input.catId,
userId: input.userId,
Expand Down Expand Up @@ -191,6 +194,7 @@ export class SessionChainStore implements ISessionChainStore {
record.cliSessionId = patch.cliSessionId;
this.cliIndex.set(patch.cliSessionId, id);
}
if (patch.workingDirectory !== undefined) record.workingDirectory = patch.workingDirectory;
if (patch.status !== undefined) {
record.status = patch.status;
const key = this.catThreadKey(record.catId, record.threadId);
Expand Down
12 changes: 11 additions & 1 deletion packages/api/src/domains/cats/services/stores/redis/RedisSessionChainStore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,8 @@ const DEFAULT_TTL_SECONDS = 0; // persistent — set >0 via env to enable expiry
* KEYS[4] = cli key, KEYS[5] = chainKey index key (F198; dummy when no chainKey)
* ARGV[1] = id, ARGV[2] = cliSessionId, ARGV[3] = threadId, ARGV[4] = catId,
* ARGV[5] = userId, ARGV[6] = now, ARGV[7] = reuseExistingCliSession flag,
* ARGV[8] = chainKey value ('' = none, KEYS[5] left untouched)
* ARGV[8] = chainKey value ('' = none, KEYS[5] left untouched),
* ARGV[9] = workingDirectory value ('' = unknown)
*
* Returns: {'existing', existingId} when cliSessionId is already claimed,
* {'created', id, seq} when a new record is created.
Expand All @@ -52,6 +53,9 @@ if ARGV[8] ~= '' then
redis.call('HSET', KEYS[3], 'chainKey', ARGV[8])
${DEFAULT_TTL_SECONDS > 0 ? `redis.call('SET', KEYS[5], ARGV[1], 'EX', ${DEFAULT_TTL_SECONDS})` : `redis.call('SET', KEYS[5], ARGV[1])`}
end
if ARGV[9] ~= '' then
redis.call('HSET', KEYS[3], 'workingDirectory', ARGV[9])
end
${DEFAULT_TTL_SECONDS > 0 ? `redis.call('EXPIRE', KEYS[3], ${DEFAULT_TTL_SECONDS})` : '-- persistent mode: no EXPIRE'}
redis.call('ZADD', KEYS[2], seq, ARGV[1])
${DEFAULT_TTL_SECONDS > 0 ? `redis.call('EXPIRE', KEYS[2], ${DEFAULT_TTL_SECONDS})` : '-- persistent mode: no EXPIRE'}
Expand Down Expand Up @@ -112,6 +116,7 @@ export class RedisSessionChainStore implements ISessionChainStore {
now,
input.reuseExistingCliSession ? '1' : '0',
input.chainKey ?? '',
input.workingDirectory ?? '',
)) as [string, string, string?];

const [status, recordId, seqRaw] = result;
Expand All @@ -126,6 +131,7 @@ export class RedisSessionChainStore implements ISessionChainStore {
return {
id: recordId,
cliSessionId: input.cliSessionId,
...(input.workingDirectory ? { workingDirectory: input.workingDirectory } : {}),
threadId: input.threadId,
catId: input.catId as CatId,
userId: input.userId,
Expand Down Expand Up @@ -229,6 +235,9 @@ export class RedisSessionChainStore implements ISessionChainStore {
}
pairs.push('cliSessionId', patch.cliSessionId);
}
if (patch.workingDirectory !== undefined) {
pairs.push('workingDirectory', patch.workingDirectory);
}

if (patch.status !== undefined) {
pairs.push('status', patch.status);
Expand Down Expand Up @@ -354,6 +363,7 @@ export class RedisSessionChainStore implements ISessionChainStore {
return {
id: data.id!,
cliSessionId: data.cliSessionId!,
...(data.workingDirectory ? { workingDirectory: data.workingDirectory } : {}),
threadId: data.threadId!,
catId: data.catId as CatId,
userId: data.userId!,
Expand Down
176 changes: 176 additions & 0 deletions packages/api/test/invoke-single-cat.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6835,6 +6835,182 @@ describe('invokeSingleCat audit events (P1 fix)', () => {
assert.equal(optionsSeen[0]?.workingDirectory, projectRoot);
});

it('does not resume OpenCode when active session has no stored workspace', async () => {
const projectRoot = await realpath(join(__dirname, '..', '..', '..'));
const optionsSeen = [];
const service = {
l0CompilerFn: dummyL0CompilerFn,
async *invoke(_prompt, options) {
optionsSeen.push(options ?? {});
yield { type: 'done', catId: 'opencode', timestamp: Date.now() };
},
};
const activeRecord = {
id: 'rec-opencode-no-workspace',
seq: 0,
status: 'active',
cliSessionId: 'stale-opencode-session',
catId: 'opencode',
threadId: 'thread-opencode-no-workspace-resume',
userId: 'user1',
messageCount: 1,
createdAt: Date.now() - 1000,
updatedAt: Date.now() - 1000,
};
const deps = {
...makeDeps(),
threadStore: {
get: async () => ({ projectPath: projectRoot, createdBy: 'user1' }),
updateParticipantActivity: async () => {},
},
sessionManager: {
get: async () => 'stale-opencode-session',
store: async () => {},
delete: async () => {},
},
sessionChainStore: {
getChain: async () => [activeRecord],
getActive: async () => activeRecord,
get: async () => activeRecord,
create: async () => activeRecord,
update: async () => activeRecord,
},
};

await collect(
invokeSingleCat(deps, {
catId: 'opencode',
service,
prompt: 'test stale workspace resume',
userId: 'user1',
threadId: 'thread-opencode-no-workspace-resume',
isLastCat: true,
}),
);

assert.equal(optionsSeen.length, 1);
assert.equal(optionsSeen[0]?.sessionId, undefined, 'OpenCode must not resume sessions without workspace metadata');
assert.equal(optionsSeen[0]?.workingDirectory, projectRoot);
});

it('does not resume OpenCode when active session workspace differs from thread workspace', async () => {
const projectRoot = await realpath(join(__dirname, '..', '..', '..'));
const optionsSeen = [];
const service = {
l0CompilerFn: dummyL0CompilerFn,
async *invoke(_prompt, options) {
optionsSeen.push(options ?? {});
yield { type: 'done', catId: 'opencode', timestamp: Date.now() };
},
};
const activeRecord = {
id: 'rec-opencode-wrong-workspace',
seq: 0,
status: 'active',
cliSessionId: 'stale-opencode-session',
workingDirectory: '/tmp/other-workspace',
catId: 'opencode',
threadId: 'thread-opencode-wrong-workspace-resume',
userId: 'user1',
messageCount: 1,
createdAt: Date.now() - 1000,
updatedAt: Date.now() - 1000,
};
const deps = {
...makeDeps(),
threadStore: {
get: async () => ({ projectPath: projectRoot, createdBy: 'user1' }),
updateParticipantActivity: async () => {},
},
sessionManager: {
get: async () => 'stale-opencode-session',
store: async () => {},
delete: async () => {},
},
sessionChainStore: {
getChain: async () => [activeRecord],
getActive: async () => activeRecord,
get: async () => activeRecord,
create: async () => activeRecord,
update: async () => activeRecord,
},
};

await collect(
invokeSingleCat(deps, {
catId: 'opencode',
service,
prompt: 'test wrong workspace resume',
userId: 'user1',
threadId: 'thread-opencode-wrong-workspace-resume',
isLastCat: true,
}),
);

assert.equal(optionsSeen.length, 1);
assert.equal(optionsSeen[0]?.sessionId, undefined, 'OpenCode must not resume sessions from a different workspace');
assert.equal(optionsSeen[0]?.workingDirectory, projectRoot);
});

it('resumes OpenCode when active session workspace matches thread workspace', async () => {
const projectRoot = await realpath(join(__dirname, '..', '..', '..'));
const optionsSeen = [];
const service = {
l0CompilerFn: dummyL0CompilerFn,
async *invoke(_prompt, options) {
optionsSeen.push(options ?? {});
yield { type: 'done', catId: 'opencode', timestamp: Date.now() };
},
};
const activeRecord = {
id: 'rec-opencode-matching-workspace',
seq: 0,
status: 'active',
cliSessionId: 'live-opencode-session',
workingDirectory: projectRoot,
catId: 'opencode',
threadId: 'thread-opencode-matching-workspace-resume',
userId: 'user1',
messageCount: 1,
createdAt: Date.now() - 1000,
updatedAt: Date.now() - 1000,
};
const deps = {
...makeDeps(),
threadStore: {
get: async () => ({ projectPath: projectRoot, createdBy: 'user1' }),
updateParticipantActivity: async () => {},
},
sessionManager: {
get: async () => 'live-opencode-session',
store: async () => {},
delete: async () => {},
},
sessionChainStore: {
getChain: async () => [activeRecord],
getActive: async () => activeRecord,
get: async () => activeRecord,
create: async () => activeRecord,
update: async () => activeRecord,
},
};

await collect(
invokeSingleCat(deps, {
catId: 'opencode',
service,
prompt: 'test matching workspace resume',
userId: 'user1',
threadId: 'thread-opencode-matching-workspace-resume',
isLastCat: true,
}),
);

assert.equal(optionsSeen.length, 1);
assert.equal(optionsSeen[0]?.sessionId, 'live-opencode-session');
assert.equal(optionsSeen[0]?.workingDirectory, projectRoot);
});

it('fails loud for OpenCode when thread projectPath is default', async () => {
let invokedService = false;
const service = {
Expand Down
Loading
Loading