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

fix(fs): observe absence before guarded recreation

Browse Source
This commit is contained in:
Tianyi Cui
2026-08-09 15:22:50 +08:00
parent 9aa2d07353
commit ceba53edd7
68 changed files with 694 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/fs/fs-policy/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/fs/fs-policy/README.md
README.md: bf6486f6f7fe574b576a71b430be91031f7805d2
README.zh.md: 2e38d63d8e310c7f22b764d46cbec0c82ffb96cf
README.md: 0166690c66f38efe817d8b5779db9a677d22d321
README.zh.md: 26e0199e6deb7fc12a860ab01884b70c1b7c9b38

12
packages/fs/fs-policy/README.md
View File

@@ -2,7 +2,7 @@
English | [中文](README.zh.md)
The **fs-policy plugin**: it adds observed-state, read-before-edit, and version-guarded write/edit on top of the `ctx.fs` provider contract ([`@deepseek-ai/dsh-fs`](../fs)) — through the `fs/*` event gate, **NOT** through a method service. This plugin registers **no** `ctx.fsPolicy` service and has no public `read`/`write`/`edit`/`resolve` methods. It is the policy third of the filesystem stack: not a swappable seam, but the policy that does not belong on the `FileSystem` provider base class.
The **fs-policy plugin**: it records observed presence or absence and adds read-before-edit plus guarded write/edit on top of the `ctx.fs` provider contract ([`@deepseek-ai/dsh-fs`](../fs)) — through the `fs/*` event gate, **NOT** through a method service. This plugin registers **no** `ctx.fsPolicy` service and has no public `read`/`write`/`edit`/`resolve` methods. It is the policy third of the filesystem stack: not a swappable seam, but the policy that does not belong on the `FileSystem` provider base class.
```ts
import type { Context } from 'cordis'
@@ -33,13 +33,13 @@ Three `fs/*` events (declared by `@deepseek-ai/dsh-fs`, dispatched by `@deepseek
| Event | This plugin's listener |
|---|---|
| `fs/write-intent` | No prior observation → `{ kind: 'createIfAbsent' }`; a prior observation → `{ kind: 'replaceIfVersion', version: vObserved }`. Single-slot decision; does NOT call `next()`. |
| `fs/edit-intent` | Requires a prior observation by this owner (else throws `FS_NOT_OBSERVED`); returns `{ version: vObserved }` as the CAS basis. Single-slot decision; does NOT call `next()`. |
| `fs/observed` | Records `{ version }` for this owner+target. Synchronous, side-effect-only `WeakMap.set`. |
| `fs/write-intent` | Unseen or observed absent → `{ kind: 'createIfAbsent' }`; observed present → `{ kind: 'replaceIfVersion', version: vObserved }`. Single-slot decision; does NOT call `next()`. |
| `fs/edit-intent` | Unseen → `FS_NOT_OBSERVED`; observed absent → `FS_NOT_FOUND`; observed present → `{ version: vObserved }` as the CAS basis. Single-slot decision; does NOT call `next()`. |
| `fs/observed` | Records `{ kind: 'present', version }` or `{ kind: 'absent' }` for this owner+target. Synchronous, side-effect-only `WeakMap.set`. |
## Observed state is the prior-observation record; freshness is provider CAS
Observed state is a weak owner-to-target version map updated after every successful read or mutation; presence alone is the prior-observation record. The plugin performs no filesystem I/O: it supplies the observed version to the provider's atomic mutation guard. A windowed read observes the whole file version, so a later targeted edit is allowed only while that file remains unchanged. State is discarded on plugin disposal and is not persisted across sessions.
Observed state is a weak owner-to-target map with three logical states: unseen, confirmed absent, or present at a version. A successful file read or mutation records presence; a `read`/`view` metadata miss records absence before returning `FS_NOT_FOUND`. The plugin performs no filesystem I/O: it converts that state into a provider guard. Presence supplies the observed version, while absence lets only a `createIfAbsent` write proceed; edit has no version basis and returns `FS_NOT_FOUND`. A windowed read observes the whole file version, so a later targeted edit is allowed only while that file remains unchanged. State is discarded on plugin disposal and is not persisted across sessions.
## Single-slot, first-wins
@@ -55,7 +55,7 @@ Because the plugin influences the world only through events, removing it does no
#### What the model sees
This plugin adds no prompt or schema. It rejects an edit without a prior read with code `FS_NOT_OBSERVED` and exact message `edit requires reading "<path>" first`. Guarded mutations whose observed version is stale propagate the provider-owned `FS_STALE_VERSION` error. [`dsh-tool-fs`](../tool-fs/README.md) owns the model-facing error wrapper, which appends the recovery instruction to `FS_STALE_VERSION` (`— re-read the file, then retry`) and `FS_NOT_OBSERVED` (`— read the file, then retry`) messages while preserving the code; observation state is never shown.
This plugin adds no prompt or schema. It rejects an edit without a prior observation with code `FS_NOT_OBSERVED` and exact message `edit requires reading "<path>" first`; editing a target just observed absent returns `FS_NOT_FOUND`. Guarded mutations whose positive observation is stale propagate the provider-owned `FS_STALE_VERSION` error. [`dsh-tool-fs`](../tool-fs/README.md) owns the model-facing error wrapper, which appends the recovery instruction to `FS_STALE_VERSION` (`— re-read the file, then retry`) and `FS_NOT_OBSERVED` (`— read the file, then retry`) messages while preserving the code. Following the stale remedy on an externally deleted target now records absence: the next guarded write may recreate it with `createIfAbsent`, while the provider atomically preserves any concurrent creator.
#### Token effect

12
packages/fs/fs-policy/README.zh.md
View File

@@ -2,7 +2,7 @@
[English](README.md) | 中文
**fs-policy 插件**：它在 `ctx.fs` 提供方约定（[`@deepseek-ai/dsh-fs`](../fs)）之上增加已观察状态、编辑前读取和版本防护的写入/编辑；它通过 `fs/*` 事件门禁参与，**不是**通过方法服务。该插件**不**注册 `ctx.fsPolicy` 服务，也没有公开的 `read`/`write`/`edit`/`resolve` 方法。它是文件系统栈的策略层：不是可替换 seam，而是不应位于 `FileSystem` 提供方基类上的策略。
**fs-policy 插件**：它记录观测到的存在或缺失状态，并在 `ctx.fs` 提供方约定（[`@deepseek-ai/dsh-fs`](../fs)）之上增加编辑前读取和带防护的写入/编辑；它通过 `fs/*` 事件门禁参与，**不是**通过方法服务。该插件**不**注册 `ctx.fsPolicy` 服务，也没有公开的 `read`/`write`/`edit`/`resolve` 方法。它是文件系统栈的政策层：不是可替换 seam，而是不应位于 `FileSystem` 提供方基类上的政策。
```ts
import type { Context } from 'cordis'
@@ -33,13 +33,13 @@ await ctx.plugin(FsPolicy)
| 事件 | 本插件的监听器 |
|---|---|
| `fs/write-intent` | 先前未观察 → `{ kind: 'createIfAbsent' }`；先前已观察 → `{ kind: 'replaceIfVersion', version: vObserved }`。单槽决策；不调用 `next()`。 |
| `fs/edit-intent` | 要求该所有者先前已观察，否则抛出 `FS_NOT_OBSERVED`；返回 `{ version: vObserved }` 作为 CAS 基础。单槽决策；不调用 `next()`。 |
| `fs/observed` | 为该所有者与目标记录 `{ version }`。同步、只有副作用的 `WeakMap.set`。 |
| `fs/write-intent` | 未见或已观测为缺失 → `{ kind: 'createIfAbsent' }`；已观测为存在 → `{ kind: 'replaceIfVersion', version: vObserved }`。单槽决策；不调用 `next()`。 |
| `fs/edit-intent` | 未见 → `FS_NOT_OBSERVED`；已观测为缺失 → `FS_NOT_FOUND`；已观测为存在 → 返回 `{ version: vObserved }` 作为 CAS 基础。单槽决策；不调用 `next()`。 |
| `fs/observed` | 为该所有者与目标记录 `{ kind: 'present', version }` 或 `{ kind: 'absent' }`。同步、只有副作用的 `WeakMap.set`。 |
## 已观察状态是先前观察记录；新鲜度由提供方 CAS 保证
已观察状态是一张以所有者为弱键、记录各目标版本的映射表，每次读取或变更成功后都会更新；记录存在本身就是先前观察凭据。插件不执行文件系统 I/O：它把观察到的版本提供给提供方的原子变更防护。窗口读取会观察整个文件的版本，因此只有文件保持不变时才允许后续的定向编辑。插件 dispose（资源释放）时会丢弃状态，并且不会跨会话持久化。
观测状态是一张以所有者为弱键、记录各目标的映射表，具有三种逻辑状态：未见、确认缺失、存在于某个版本。成功读取文件或变更会记录存在；`read`/`view` 的元数据未命中会在返回 `FS_NOT_FOUND` 前记录缺失。插件不执行文件系统 I/O：它把该状态转换为提供方防护。存在状态提供观测到的版本；缺失状态只允许 `createIfAbsent` 写入继续，edit 因没有版本基准而返回 `FS_NOT_FOUND`。窗口读取会观察整个文件的版本，因此只有文件保持不变时才允许后续的定向编辑。插件 dispose（资源释放）时会丢弃状态，并且不会跨会话持久化。
## 单槽、先到者胜
@@ -55,7 +55,7 @@ await ctx.plugin(FsPolicy)
#### 模型看到的内容
该插件不添加提示词或 schema。编辑前未读取时，它会以代码 `FS_NOT_OBSERVED` 和精确消息 `edit requires reading "<path>" first` 拒绝。观察版本陈旧的防护变更会传播由提供方拥有的 `FS_STALE_VERSION` 错误。[`dsh-tool-fs`](../tool-fs/README.md)拥有面向模型的错误包装，会为 `FS_STALE_VERSION` 消息追加恢复指令（`— re-read the file, then retry`）、为 `FS_NOT_OBSERVED` 消息追加恢复指令（`— read the file, then retry`），同时保留错误码；观察状态绝不会显示。
该插件不添加提示词或 schema。没有先前观测时，它会以代码 `FS_NOT_OBSERVED` 和精确消息 `edit requires reading "<path>" first` 拒绝编辑；编辑刚被观测为缺失的目标会返回 `FS_NOT_FOUND`。正向观测陈旧时，带防护的变更会传播由提供方拥有的 `FS_STALE_VERSION` 错误。[`dsh-tool-fs`](../tool-fs/README.md)拥有面向模型的错误包装，会为 `FS_STALE_VERSION` 消息追加恢复指令（`— re-read the file, then retry`）、为 `FS_NOT_OBSERVED` 消息追加恢复指令（`— read the file, then retry`），同时保留错误码。外部删除目标后，遵循陈旧恢复指令会记录缺失：下一次带防护的写入可以通过 `createIfAbsent` 重新创建该目标，而提供方会以原子方式保留任何并发创建者写入的文件。
#### Token 影响

56
packages/fs/fs-policy/src/index.ts
View File

@@ -1,14 +1,15 @@
/**
* Event-only filesystem observation policy; it registers no service. A weak owner/target map
* records every successful read or mutation, single-slot intent listeners supply that version,
* and the provider performs the atomic freshness check. Without this plugin, tools retain the
* bare provider's unconditional mutation behavior. See the package README for composition rules.
* records every authoritative presence/absence observation, single-slot intent listeners derive
* guards from that state, and the provider performs the atomic freshness/no-clobber check. Without
* this plugin, tools retain the bare provider's unconditional mutation behavior. See the package
* README for composition rules.
* @module @deepseek-ai/dsh-fs-policy
*/
import type { Context } from 'cordis'
import { FsError } from '@deepseek-ai/dsh-fs'
import type { FsTarget, FsVersion, FsWriteIntent } from '@deepseek-ai/dsh-fs'
import type { FsObservation, FsTarget, FsVersion, FsWriteIntent } from '@deepseek-ai/dsh-fs'
import type { FsPolicyExec } from './types.ts'
export type { FsPolicyExec } from './types.ts'
@@ -21,9 +22,10 @@ class ObservedStateGate {
/**
* Observed-file state, keyed first by the owner object (weakly held, so a
* collected session frees its state), then by {@link FsTarget.targetKey}. An
* entry's PRESENCE is the prior-observation record.
* entry's presence is the prior-observation record; its discriminant keeps
* confirmed absence distinct from an unseen target.
*/
private observed = new WeakMap<object, Map<string, FsVersion>>()
private observed = new WeakMap<object, Map<string, FsObservation>>()
/**
* Derive the observed-state owner from the opaque event actor — normally the
@@ -38,17 +40,17 @@ class ObservedStateGate {
return (actor as FsPolicyExec | undefined)?.agent?.session
}
private get(owner: object, targetKey: string): FsVersion | undefined {
private get(owner: object, targetKey: string): FsObservation | undefined {
return this.observed.get(owner)?.get(targetKey)
}
private set(owner: object, targetKey: string, version: FsVersion): void {
private set(owner: object, targetKey: string, observation: FsObservation): void {
let byTarget = this.observed.get(owner)
if (!byTarget) {
byTarget = new Map()
this.observed.set(owner, byTarget)
}
byTarget.set(targetKey, version)
byTarget.set(targetKey, observation)
}
/** Drop all recorded state (HMR safety / disposal). */
@@ -57,33 +59,38 @@ class ObservedStateGate {
}
/**
* Decide the write intent: no prior observation ⇒ `createIfAbsent` (only
* new files can be created blindly); a prior observation ⇒ `replaceIfVersion`
* at the observed version (existing files replaced only if unchanged).
* Decide the write intent: unseen or confirmed absent ⇒ `createIfAbsent`;
* confirmed present ⇒ `replaceIfVersion` at the observed version.
*/
writeIntent(target: FsTarget, actor: object | undefined): FsWriteIntent {
const owner = this.owner(actor)
const prior = owner ? this.get(owner, target.targetKey) : undefined
return prior ? { kind: 'replaceIfVersion', version: prior } : { kind: 'createIfAbsent' }
return prior?.kind === 'present'
? { kind: 'replaceIfVersion', version: prior.version }
: { kind: 'createIfAbsent' }
}
/**
* Decide the edit version guard: requires a prior observation by this owner
* (else `FS_NOT_OBSERVED`); returns the observed version as the CAS basis.
* Decide the edit version guard: unseen rejects with `FS_NOT_OBSERVED`,
* confirmed absence rejects with `FS_NOT_FOUND`, and presence supplies the
* observed version as the CAS basis.
*/
editIntent(target: FsTarget, actor: object | undefined): { version: FsVersion } {
const owner = this.owner(actor)
const prior = owner ? this.get(owner, target.targetKey) : undefined
if (!owner || !prior) {
if (!owner || prior === undefined) {
throw new FsError(`edit requires reading "${target.displayPath}" first`, 'FS_NOT_OBSERVED')
}
return { version: prior }
if (prior.kind === 'absent') {
throw new FsError(`cannot edit "${target.displayPath}": not found`, 'FS_NOT_FOUND')
}
return { version: prior.version }
}
/** Record a successful read/write/edit: this owner observed this target at this version. */
observe(target: FsTarget, version: FsVersion, actor: object | undefined): void {
/** Record an authoritative present or absent observation for this owner and target. */
observe(target: FsTarget, observation: FsObservation, actor: object | undefined): void {
const owner = this.owner(actor)
if (owner) this.set(owner, target.targetKey, version)
if (owner) this.set(owner, target.targetKey, observation)
}
}
@@ -114,9 +121,10 @@ export function apply(ctx: Context): void {
// fs/edit-intent: occupy the single decision slot — do not call next().
ctx.on('fs/edit-intent', (target, actor) => Promise.resolve().then(() => gate.editIntent(target, actor)))
// fs/observed must remain synchronous and non-throwing: the mutation already succeeded, and
// emit does not await promises. WeakMap.set satisfies that contract.
ctx.on('fs/observed', (target, version, actor) => {
gate.observe(target, version, actor)
// fs/observed must remain synchronous and non-throwing: emit does not await
// promises, and successful mutations have already committed. WeakMap.set
// satisfies that contract for both presence and absence.
ctx.on('fs/observed', (target, observation, actor) => {
gate.observe(target, observation, actor)
})
}

53
packages/fs/fs-policy/tests/policy.spec.ts
View File

@@ -3,7 +3,7 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { FsTargetKey, FsVersion } from '@deepseek-ai/dsh-fs'
import type { FsTarget, FsWriteIntent } from '@deepseek-ai/dsh-fs'
import type { FsObservation, FsTarget, FsWriteIntent } from '@deepseek-ai/dsh-fs'
import * as FsPolicy from '@deepseek-ai/dsh-fs-policy'
import type { FsPolicyExec } from '@deepseek-ai/dsh-fs-policy'
@@ -11,6 +11,8 @@ function target(path: string): FsTarget {
return { targetKey: FsTargetKey(path), displayPath: path }
}
const ownerExec = (session: object): FsPolicyExec => ({ agent: { session } })
const present = (version: string): FsObservation => ({ kind: 'present', version: FsVersion(version) })
const absent: FsObservation = { kind: 'absent' }
/** Dispatch the write-intent waterfall with the bare default thunk. */
function writeIntent(ctx: Context, t: FsTarget, actor: object | undefined): Promise<FsWriteIntent | undefined> {
@@ -64,9 +66,16 @@ describe('write-intent decision', () => {
it('an observed target decides replaceIfVersion at the observed version', async () => {
const { ctx } = await setup()
const exec = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), FsVersion('v7'), exec)
ctx.emit('fs/observed', target('a.txt'), present('v7'), exec)
expect(await writeIntent(ctx, target('a.txt'), exec)).toEqual({ kind: 'replaceIfVersion', version: 'v7' })
})
it('a target observed absent decides createIfAbsent', async () => {
const { ctx } = await setup()
const exec = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), absent, exec)
expect(await writeIntent(ctx, target('a.txt'), exec)).toEqual({ kind: 'createIfAbsent' })
})
})
describe('edit-intent decision', () => {
@@ -88,16 +97,23 @@ describe('edit-intent decision', () => {
it('returns the observed version as the CAS basis after an observation', async () => {
const { ctx } = await setup()
const exec = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), FsVersion('v3'), exec)
ctx.emit('fs/observed', target('a.txt'), present('v3'), exec)
expect(await editIntent(ctx, target('a.txt'), exec)).toEqual({ version: 'v3' })
})
it('rejects editing a target observed absent with FS_NOT_FOUND', async () => {
const { ctx } = await setup()
const exec = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), absent, exec)
await expect(editIntent(ctx, target('a.txt'), exec)).rejects.toMatchObject({ code: 'FS_NOT_FOUND' })
})
})
describe('observed-state is the prior-observation record', () => {
it('a read observation authorizes an in-place write at that version', async () => {
const { ctx } = await setup()
const exec = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), FsVersion('v0'), exec) // a read
ctx.emit('fs/observed', target('a.txt'), present('v0'), exec) // a read
expect(await writeIntent(ctx, target('a.txt'), exec)).toEqual({ kind: 'replaceIfVersion', version: 'v0' })
})
@@ -105,19 +121,34 @@ describe('observed-state is the prior-observation record', () => {
const { ctx } = await setup()
const exec = ownerExec({})
// A create records v1; the follow-up edit guards against v1 with no read.
ctx.emit('fs/observed', target('a.txt'), FsVersion('v1'), exec)
ctx.emit('fs/observed', target('a.txt'), present('v1'), exec)
expect(await editIntent(ctx, target('a.txt'), exec)).toEqual({ version: 'v1' })
// The edit records v2; a second edit guards against v2.
ctx.emit('fs/observed', target('a.txt'), FsVersion('v2'), exec)
ctx.emit('fs/observed', target('a.txt'), present('v2'), exec)
expect(await editIntent(ctx, target('a.txt'), exec)).toEqual({ version: 'v2' })
})
it('a no-owner observation records nothing', async () => {
const { ctx } = await setup()
ctx.emit('fs/observed', target('a.txt'), FsVersion('v0'), undefined)
ctx.emit('fs/observed', target('a.txt'), present('v0'), undefined)
// Still unobserved for any owner.
await expect(editIntent(ctx, target('a.txt'), ownerExec({}))).rejects.toMatchObject({ code: 'FS_NOT_OBSERVED' })
})
it('supports present → absent → present transitions for one owner', async () => {
const { ctx } = await setup()
const exec = ownerExec({})
const a = target('a.txt')
ctx.emit('fs/observed', a, present('v1'), exec)
expect(await writeIntent(ctx, a, exec)).toEqual({ kind: 'replaceIfVersion', version: 'v1' })
ctx.emit('fs/observed', a, absent, exec)
expect(await writeIntent(ctx, a, exec)).toEqual({ kind: 'createIfAbsent' })
await expect(editIntent(ctx, a, exec)).rejects.toMatchObject({ code: 'FS_NOT_FOUND' })
ctx.emit('fs/observed', a, present('v2'), exec)
expect(await editIntent(ctx, a, exec)).toEqual({ version: 'v2' })
})
})
describe('multi-owner isolation', () => {
@@ -125,7 +156,7 @@ describe('multi-owner isolation', () => {
const { ctx } = await setup()
const a = ownerExec({})
const b = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), FsVersion('v0'), a)
ctx.emit('fs/observed', target('a.txt'), present('v0'), a)
await expect(editIntent(ctx, target('a.txt'), b)).rejects.toMatchObject({ code: 'FS_NOT_OBSERVED' })
expect(await editIntent(ctx, target('a.txt'), a)).toEqual({ version: 'v0' })
})
@@ -134,7 +165,7 @@ describe('multi-owner isolation', () => {
const { ctx } = await setup()
const a = ownerExec({})
const b = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), FsVersion('v0'), a) // A observed v0
ctx.emit('fs/observed', target('a.txt'), present('v0'), a) // A observed v0
// B never observed → createIfAbsent; A still holds v0 → replaceIfVersion.
expect(await writeIntent(ctx, target('a.txt'), b)).toEqual({ kind: 'createIfAbsent' })
expect(await writeIntent(ctx, target('a.txt'), a)).toEqual({ kind: 'replaceIfVersion', version: 'v0' })
@@ -164,7 +195,7 @@ describe('single-slot, first-wins', () => {
return Promise.resolve(undefined)
})
const exec = ownerExec({})
ctx.emit('fs/observed', target('a.txt'), FsVersion('v0'), exec)
ctx.emit('fs/observed', target('a.txt'), present('v0'), exec)
await editIntent(ctx, target('a.txt'), exec)
expect(secondRan).toBe(false)
})
@@ -186,7 +217,7 @@ describe('disposal releases recorded state (HMR safety)', () => {
const ctx = new Context()
const exec = ownerExec({})
const fiber = await ctx.plugin(FsPolicy)
ctx.emit('fs/observed', target('a.txt'), FsVersion('v0'), exec)
ctx.emit('fs/observed', target('a.txt'), present('v0'), exec)
expect(await editIntent(ctx, target('a.txt'), exec)).toEqual({ version: 'v0' })
await fiber.dispose()