Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(workspace): make deletion recoverable

Browse Source
This commit is contained in:
NI0317
2026-07-27 14:37:04 +08:00
parent 187cf6f804
commit 7bd96af5eb
13 changed files with 294 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/workspace/workspace/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/workspace/workspace/README.md
README.md: 52d03b33b3482dcb6a2f5feddbc15ac9fefee0a8
README.zh.md: f899abdc3dd2a551179cd710c6dda84f804a8e80
README.md: bee3e4fcb5dded273f30942ee2e42ee93b839e62
README.zh.md: 7960a2d13df4f881687fd88cdb07e237b3abb7c8

2
packages/workspace/workspace/README.md
View File

@@ -18,6 +18,8 @@ The entity/storage rationale lives in the [domain Agent Note](../../../.agents/n
`storageDomain` and `sessionPersistence` are required startup dependencies. An unavailable peer leaves the plugin pending and cannot commit an empty initialized marker. On the first successful start, the registry calls `SessionPersistence.list()` and uses only header `id`, `cwd`, and `createdAt` to group valid historical directories and persist initial order; it never reads event bodies. The initialized marker is written last, so partial bootstrap writes are reused safely after restart. Later cwd-only sessions remain Ungrouped.
Create and delete persist an explicit pending-mutation marker before their record and order can diverge. Startup completes only the marked mutation, then clears the marker; an unmarked order/table mismatch remains unexplained corruption and fails loud. Deleting and re-registering the same path creates a fresh Workspace id and does not automatically re-adopt the retained Sessions.
## Model Experience
### Workspace records and session accounts

2
packages/workspace/workspace/README.zh.md
View File

@@ -18,6 +18,8 @@ DeepSeek Harness 的 Workspace 实体注册表（`ctx.workspace`）：通过领
`storageDomain` 和 `sessionPersistence` 是启动必需依赖。对等服务不可用时，插件保持待处理，且不能提交空的已初始化标记。首次成功启动时，注册表调用 `SessionPersistence.list()`，仅使用头部 `id`、`cwd` 和 `createdAt` 对有效历史目录分组并持久化初始顺序；它绝不读取事件正文。已初始化标记最后写入，因此重启后可安全复用部分启动写入。后续仅有 cwd 的会话仍属于 Ungrouped。
Create 与 delete 会在记录和顺序可能分叉之前，先持久化明确的待处理变更标记。启动时只补全被该标记证明的变更，随后清除标记；没有标记的顺序／表不一致仍属于来源不明的损坏，并会直接失败。删除后重新注册同一路径会生成新的 Workspace id，且不会自动重新接纳保留下来的 Session。
## 模型体验
### Workspace 记录与会话记账

75
packages/workspace/workspace/src/index.ts
View File

@@ -109,6 +109,7 @@ export class WorkspaceRegistry extends Service {
this.global = domain.global
this.state = domain.global.get()
await this.recoverPendingMutation()
this.validateStoredState(this.state)
if (!this.state.initialized) {
const headers = await this.ctx.sessionPersistence.list()
@@ -218,10 +219,28 @@ export class WorkspaceRegistry extends Service {
}
const entity = new WorkspaceEntity(this.host, id, record)
this.entities.set(id, entity)
const pendingState: WorkspaceDomainState = {
...state,
pendingMutation: { operation: 'create', workspaceId: id },
}
try {
await this.setState(pendingState)
} catch (error) {
this.entities.delete(id)
throw error
}
try {
await table.put(id, record)
} catch (error) {
this.entities.delete(id)
try {
await this.setState(state)
} catch (rollbackError) {
throw new AggregateError(
[error, rollbackError],
`workspace '${id}' record write and pending-marker rollback both failed`,
)
}
throw error
}
@@ -232,10 +251,17 @@ export class WorkspaceRegistry extends Service {
try {
await table.delete(id)
} catch (rollbackError) {
this.entities.set(id, entity)
throw new AggregateError(
[error, rollbackError],
`workspace '${id}' was stored but its registry order and rollback both failed`,
`workspace '${id}' order write and record rollback both failed; the pending marker remains recoverable`,
)
}
try {
await this.setState(state)
} catch (rollbackError) {
throw new AggregateError(
[error, rollbackError],
`workspace '${id}' order write and pending-marker rollback both failed`,
)
}
throw error
@@ -251,7 +277,10 @@ export class WorkspaceRegistry extends Service {
initialized: true,
workspaceIds: state.workspaceIds.filter(workspaceId => workspaceId !== id),
}
await this.setState(nextState)
await this.setState({
...nextState,
pendingMutation: { operation: 'delete', workspaceId: id },
})
this.entities.delete(id)
try {
await this.requireTable().delete(id)
@@ -260,6 +289,10 @@ export class WorkspaceRegistry extends Service {
try {
await this.setState(state)
} catch (rollbackError) {
// The durable marker still says to finish deletion, so the cache must
// agree with that recoverable direction rather than republish a row
// absent from the persisted order.
this.entities.delete(id)
throw new AggregateError(
[error, rollbackError],
`workspace '${id}' record deletion and registry-order rollback both failed`,
@@ -267,9 +300,38 @@ export class WorkspaceRegistry extends Service {
}
throw error
}
try {
await this.setState(nextState)
} catch (error) {
// The deletion committed at the table write and was already published
// to Host streams. Keep the durable marker for startup recovery rather
// than reporting failure after the requested state became true.
this.ctx.logger.warn(
`workspace '${id}' was deleted but its pending marker could not be cleared: ${String(error)}`,
)
}
return true
}
/**
* Complete the one mutation explicitly named by durable state. Unexplained
* order/table divergence still reaches {@link validateStoredState} and
* fails loud; this path never infers provenance from shape alone.
*/
private async recoverPendingMutation(): Promise<void> {
const state = this.requireState()
const pending = state.pendingMutation
if (pending === undefined) return
if (state.workspaceIds.includes(pending.workspaceId)) {
throw new Error(
`workspace domain is inconsistent: pending ${pending.operation} workspace `
+ `'${pending.workspaceId}' is still present in registry order`,
)
}
await this.requireTable().delete(pending.workspaceId)
await this.setState({ initialized: state.initialized, workspaceIds: state.workspaceIds })
}
private async bootstrap(headers: readonly SessionHeader[]): Promise<void> {
const table = this.requireTable()
const state = this.requireState()
@@ -493,7 +555,12 @@ export class WorkspaceRegistry extends Service {
}
private enqueueOperation<T>(operation: () => Promise<T>): Promise<T> {
const result = this.operationTail.then(operation)
const result = this.operationTail.then(async () => {
// A committed delete may leave only its marker cleanup pending. Retry
// recovery before another create/delete can overwrite that provenance.
await this.recoverPendingMutation()
return await operation()
})
this.operationTail = result.then(() => {}, () => {})
return result
}

11
packages/workspace/workspace/src/spec.ts
View File

@@ -29,6 +29,16 @@ export const workspaceRecord = z.object({
/** One stored workspace record, inferred from {@link workspaceRecord}. */
export type WorkspaceRecord = z.infer<typeof workspaceRecord>
/**
* Recoverable two-write mutation marker. The marker is persisted before the
* record/order pair can diverge, so startup can distinguish an interrupted
* registry operation from unexplained medium corruption.
*/
const workspacePendingMutation = z.discriminatedUnion('operation', [
z.object({ operation: z.literal('create'), workspaceId }),
z.object({ operation: z.literal('delete'), workspaceId }),
])
/**
* Durable registry state. `initialized` distinguishes a valid empty registry
* from one that still needs the header-only history bootstrap;
@@ -37,6 +47,7 @@ export type WorkspaceRecord = z.infer<typeof workspaceRecord>
export const workspaceDomainState = z.object({
initialized: z.boolean(),
workspaceIds: z.array(workspaceId),
pendingMutation: workspacePendingMutation.optional(),
})
/** Durable registry state inferred from {@link workspaceDomainState}. */

151
packages/workspace/workspace/tests/workspace.spec.ts
View File

@@ -89,7 +89,7 @@ async function storageContext(pool: MemoryMediaPool, backend: StorageBackend = n
/** Backend wrapper that injects one selected bootstrap write failure. */
function selectiveFailureBackend(
pool: MemoryMediaPool,
failure: { putAt?: number; deleteAt?: number; globalAt?: number },
failure: { putAt?: number; deleteAt?: number; globalAt?: number | readonly number[] },
): StorageBackend {
const inner = new MemoryStorageBackend(pool)
let puts = 0
@@ -113,7 +113,8 @@ function selectiveFailureBackend(
},
setGlobal: async (value) => {
globals += 1
if (globals === failure.globalAt) throw new Error('selected bootstrap marker failure')
const failAt = Array.isArray(failure.globalAt) ? failure.globalAt : [failure.globalAt]
if (failAt.includes(globals)) throw new Error('selected bootstrap marker failure')
await unit.setGlobal(value)
},
close: () => unit.close(),
@@ -394,19 +395,34 @@ describe('WorkspaceRegistry create and lookup', () => {
it('rolls back the provisional cache when the record write fails', async () => {
const dir = await makeDir('write-failure')
const result = await harness()
result.pool.failNextWrites = 1
await expect(result.registry.create(dir)).rejects.toThrow(/injected/)
const pool = new MemoryMediaPool()
const result = await harness({
pool,
backend: selectiveFailureBackend(pool, { putAt: 1 }),
})
await expect(result.registry.create(dir)).rejects.toThrow(/selected bootstrap put failure/)
expect(result.registry.list()).toEqual([])
expect(await result.registry.create(dir)).toBeDefined()
})
it('does not publish a Workspace when its pending marker cannot be written', async () => {
const dir = await makeDir('pending-marker-write-failure')
const pool = new MemoryMediaPool()
const result = await harness({
pool,
backend: selectiveFailureBackend(pool, { globalAt: 2 }),
})
await expect(result.registry.create(dir)).rejects.toThrow(/selected bootstrap marker failure/)
expect(result.registry.list()).toEqual([])
expect(pool.media.get('workspace')!.tables.get('workspaces')?.size ?? 0).toBe(0)
})
it('rolls back a record when registry-order persistence fails', async () => {
const dir = await makeDir('order-write-failure')
const pool = new MemoryMediaPool()
const result = await harness({
pool,
backend: selectiveFailureBackend(pool, { globalAt: 2 }),
backend: selectiveFailureBackend(pool, { globalAt: 3 }),
})
await expect(result.registry.create(dir)).rejects.toThrow(/marker failure/)
expect(result.registry.list()).toEqual([])
@@ -418,12 +434,38 @@ describe('WorkspaceRegistry create and lookup', () => {
const pool = new MemoryMediaPool()
const result = await harness({
pool,
backend: selectiveFailureBackend(pool, { globalAt: 2, deleteAt: 1 }),
backend: selectiveFailureBackend(pool, { globalAt: 3, deleteAt: 1 }),
})
await expect(result.registry.create(dir)).rejects.toBeInstanceOf(AggregateError)
expect(pool.media.get('workspace')!.tables.get('workspaces')!.size).toBe(1)
})
it('reports a record write and pending-marker rollback failure together', async () => {
const dir = await makeDir('record-marker-rollback-failure')
const pool = new MemoryMediaPool()
const result = await harness({
pool,
backend: selectiveFailureBackend(pool, { putAt: 1, globalAt: 3 }),
})
await expect(result.registry.create(dir)).rejects.toBeInstanceOf(AggregateError)
expect(storedState(pool)).toMatchObject({
pendingMutation: { operation: 'create' },
})
})
it('reports an order write and pending-marker rollback failure together', async () => {
const dir = await makeDir('order-marker-rollback-failure')
const pool = new MemoryMediaPool()
const result = await harness({
pool,
backend: selectiveFailureBackend(pool, { globalAt: [3, 4] }),
})
await expect(result.registry.create(dir)).rejects.toBeInstanceOf(AggregateError)
expect(storedState(pool)).toMatchObject({
pendingMutation: { operation: 'create' },
})
})
it('deletes only the registration and leaves its directory and session headers untouched', async () => {
const dir = await makeDir('delete-registration')
const result = await harness({ sessions: [header('kept-session', dir)] })
@@ -440,6 +482,11 @@ describe('WorkspaceRegistry create and lookup', () => {
expect(result.list).toHaveBeenCalledTimes(1)
expect(result.load).not.toHaveBeenCalled()
expect(result.inspect).not.toHaveBeenCalled()
const reregistered = await result.registry.create(dir)
expect(reregistered.id).not.toBe(workspace.id)
expect(reregistered.path).toBe(dir)
expect(reregistered.sessionIds).toEqual([])
})
it('rolls registry order and cache back when record deletion fails', async () => {
@@ -458,11 +505,58 @@ describe('WorkspaceRegistry create and lookup', () => {
expect(storedRecord(pool, workspace.id)).toMatchObject({ path: dir })
})
it('commits deletion and leaves a recoverable marker when marker cleanup fails', async () => {
const dir = await makeDir('delete-marker-cleanup')
const pool = new MemoryMediaPool()
const first = await harness({
pool,
backend: selectiveFailureBackend(pool, { globalAt: 5 }),
})
const workspace = await first.registry.create(dir)
await expect(first.registry.delete(workspace.id)).resolves.toBe(true)
expect(first.registry.list()).toEqual([])
expect(storedState(pool)).toEqual({
initialized: true,
workspaceIds: [],
pendingMutation: { operation: 'delete', workspaceId: workspace.id },
})
const reregistered = await first.registry.create(dir)
expect(reregistered.id).not.toBe(workspace.id)
expect(storedState(pool)).toEqual({
initialized: true,
workspaceIds: [reregistered.id],
})
await first.fiber.dispose()
const restarted = await harness({ pool })
expect(restarted.registry.list().map(item => item.id)).toEqual([reregistered.id])
})
it('keeps the failed deletion unpublished when record and order rollback both fail', async () => {
const dir = await makeDir('delete-double-failure')
const pool = new MemoryMediaPool()
const result = await harness({
pool,
backend: selectiveFailureBackend(pool, { deleteAt: 1, globalAt: 5 }),
})
const workspace = await result.registry.create(dir)
await expect(result.registry.delete(workspace.id)).rejects.toBeInstanceOf(AggregateError)
expect(result.registry.get(workspace.id)).toBeUndefined()
expect(storedState(pool)).toMatchObject({
workspaceIds: [],
pendingMutation: { operation: 'delete', workspaceId: workspace.id },
})
})
it('rejects table access before the registry has started', async () => {
const dir = await makeDir('unstarted')
const registry = new WorkspaceRegistry(new Context())
await expect(registry.create(dir)).rejects.toThrow(/not started/)
expect(() => registry.list()).toThrow(/not started/)
const internals = registry as unknown as { requireTable(): unknown }
expect(() => internals.requireTable()).toThrow(/not started/)
})
})
@@ -650,6 +744,49 @@ describe('header-validated membership projection', () => {
internals.entities.delete(workspace.id)
expect(() => result.registry.list()).toThrow(/references missing workspace/)
})
it('recovers only an explicitly marked interrupted create or delete', async () => {
const createDir = await makeDir('pending-create')
const deleteDir = await makeDir('pending-delete')
const createId = WorkspaceId('00000000-0000-4000-8000-000000000004')
const deleteId = WorkspaceId('00000000-0000-4000-8000-000000000005')
const interruptedCreate = storedPool(
[[createId, record(createDir, [])]],
{
initialized: true,
workspaceIds: [],
pendingMutation: { operation: 'create', workspaceId: createId },
},
)
const createRecovery = await harness({ pool: interruptedCreate })
expect(createRecovery.registry.list()).toEqual([])
expect(interruptedCreate.media.get('workspace')!.tables.get('workspaces')!.has(createId)).toBe(false)
expect(storedState(interruptedCreate)).toEqual({ initialized: true, workspaceIds: [] })
const interruptedDelete = storedPool(
[[deleteId, record(deleteDir, [])]],
{
initialized: true,
workspaceIds: [],
pendingMutation: { operation: 'delete', workspaceId: deleteId },
},
)
const deleteRecovery = await harness({ pool: interruptedDelete })
expect(deleteRecovery.registry.list()).toEqual([])
expect(interruptedDelete.media.get('workspace')!.tables.get('workspaces')!.has(deleteId)).toBe(false)
expect(storedState(interruptedDelete)).toEqual({ initialized: true, workspaceIds: [] })
const corruptPending = storedPool(
[[deleteId, record(deleteDir, [])]],
{
initialized: true,
workspaceIds: [deleteId],
pendingMutation: { operation: 'delete', workspaceId: deleteId },
},
)
await expect(harness({ pool: corruptPending })).rejects.toThrow(/still present in registry order/)
})
})
describe('workspace mutation and status', () => {