Skip to content

Commit 99211bd

Browse files
reoringclaude
reoring
and
claude
committed
feat(workflow): add management agent with IPC protocol extensions
Add workflow management agent system including directive parsing, decision resolution, controller, and execution engine. Extend IPC protocol and core runtime to support management capabilities. Includes Effect-based structured concurrency utilities and comprehensive tests. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent f015460 commit 99211bd

45 files changed

Lines changed: 10237 additions & 100 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎bun.lock‎

Lines changed: 9 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎docs/features/workflow-management-agent.md‎

Lines changed: 1684 additions & 0 deletions
Large diffs are not rendered by default.

‎docs/issues/0010-effect-adoption-structured-concurrency.ja.md‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
# Effect 導入（structured concurrency）で Promise.race / AbortSignal / timer の複雑さを減らす
2+
3+
Status: proposal（段階導入。API破壊的な全面書き換えはしない）
4+
5+
## 課題感（いま何がつらいか）
6+
7+
Roboppi はキャンセル・タイムアウト・並行実行を、主に以下の素朴な組み合わせで実装しています。
8+
9+
- `AbortController` / `AbortSignal` の listener
10+
- `setTimeout` / `clearTimeout`
11+
- `Promise.race` と sentinel 値（`null` など）
12+
- `finally` に散らばる後始末（listener / timer / Map の掃除）
13+
14+
現時点でも動いてはいる一方で、実装が増えるほど次の問題が起きやすいです。
15+
16+
1) パターンの重複と微妙な差分
17+
- `waitForAbort`, `sleep(ms, signal)`, `createScopedAbort` などが局所最適で増殖しやすい
18+
19+
2) 後始末の正しさをレビューで担保しにくい
20+
- 早期 return / 例外 / abort / timeout の全経路で listener/timer を外す必要がある
21+
- waiter Map / request Map のリークは「後から分岐が増えたとき」に入りやすい
22+
23+
3) エラー表現が一貫しない
24+
- timeout を例外で表す箇所と、`null` / `"timeout"` などで表す箇所が混在する
25+
- `p.catch(() => {})` のような「未await時の unhandled 回避」が必要になる
26+
27+
4) 並行実行の意味がコードから読み取りにくい
28+
- 親の abort を子に伝播するのが都度手組み
29+
- submit_job -> permit -> execute -> cancel のような複合フローは race と timeout が絡みやすい
30+
31+
特に複雑さが出ている箇所（例）:
32+
33+
- `src/workflow/core-ipc-step-runner.ts`（races + scoped abort + cancel best-effort）
34+
- `src/ipc/protocol.ts`（`pendingRequests` + timer + transport close 時の fast-fail）
35+
- `src/worker/process-manager.ts`（graceful shutdown の多段 race/timeout）
36+
- `src/scheduler/supervisor.ts`（spawn/bridge 周りの `new Promise` ブロックが多い）
37+
38+
## 目的
39+
40+
1) キャンセル・タイムアウトの意味を明示し、合成可能にする
41+
2) リソース解放（listener/timer/Map cleanup）を構造化して一箇所に寄せる
42+
3) `Promise.race` / timer / listener のボイラープレートを減らし、リーク確率を下げる
43+
4) 外部挙動は維持する（公開APIは当面 Promise を返してよい）
44+
5) Bun + ESM 前提で無理なく運用する
45+
46+
## 方針（Effect-TS を内部プリミティブとして段階導入）
47+
48+
`effect` (Effect-TS) の強みを「内部実装」に限定して取り込みます。
49+
50+
- 中断可能な async 境界: `Effect.async((resume, signal) => ...)`
51+
- タイムアウト: `Effect.timeout` / `Effect.timeoutFail`
52+
- race: `Effect.race`
53+
- finalizer による確実な後始末: `Effect.acquireRelease`（Scope）
54+
- リトライ/バックオフ: `Effect.retry` + `Schedule`（指数 + jitter）
55+
- structured concurrency: fibers（`Effect.fork`, `Fiber.interrupt`, `Effect.all/forEach` + concurrency）
56+
57+
重要: 全面 rewrite せず、境界で `Effect.runPromise` して `async/await` の呼び出し方は維持します。
58+
59+
## 移行計画（段階）
60+
61+
### Phase 0: 依存追加と最小ブリッジ
62+
63+
- `bun add effect`
64+
- CI で `make typecheck` / `make test` を通す
65+
- 乱用を避けるため、最小限の内部ヘルパー層（名前 TBD）を用意する
66+
- `runPromise` ラッパ
67+
- `sleep`, `waitForAbort`, `withTimeout`, `raceAbort` など
68+
69+
### Phase 1: IPC の waiter（最もレバレッジが大きい）
70+
71+
対象: `src/ipc/protocol.ts` の `waitForResponse()`.
72+
73+
狙い:
74+
75+
- Map 登録・解除、timer の set/clear を finalizer に集約
76+
- transport close で pending を即 fail（待ち続けない）
77+
- `p.catch(() => {})` のような抑制を不要にする（または局所化する）
78+
79+
外部形は維持:
80+
81+
- API は引き続き `Promise<unknown>`
82+
- 内部だけ Effect 化し、最後に `Effect.runPromise`
83+
84+
### Phase 2: Core IPC step の複合フロー整理
85+
86+
対象: `src/workflow/core-ipc-step-runner.ts`.
87+
88+
狙い:
89+
90+
- `Promise.race([...])` の散在を減らし、abort/timeout を interruption/timeout で統一
91+
- listener cleanup を 1 箇所に寄せる
92+
- "best-effort cancel -> 最大 5s 待つ" のような手続きブロックを明示的にする
93+
94+
### Phase 3: プロセス管理（graceful shutdown）
95+
96+
対象: `src/worker/process-manager.ts`.
97+
98+
- abort listener / timeout を `acquireRelease` + `timeoutFail` で表現
99+
- SIGTERM -> SIGKILL -> stuck fallback の多段を読みやすくする
100+
101+
### Phase 4: Supervisor の spawn / transport bridging
102+
103+
対象: `src/scheduler/supervisor.ts`.
104+
105+
- callback ベース（stdin write / server close 等）を `Effect.async` で包む
106+
- temp dir / server / socket の掃除を finalizer に統一
107+
108+
### Phase 5（任意）: workflow レベルの structured concurrency
109+
110+
対象: `src/workflow/executor.ts`.
111+
112+
ここは効果も大きいが変更幅が最大。
113+
114+
- step 実行を fiber 化し、`Effect.forEach(..., { concurrency })` 等で上限を表現
115+
- workflow abort で子 fiber を一括 interrupt
116+
- `ExecEventSink` のイベント順序・意味は維持する
117+
118+
## 受け入れ条件
119+
120+
1) `make typecheck` が通る
121+
2) `make test` が通る
122+
3) キャンセル挙動の回帰がない
123+
- 親 abort で待ちが止まり、listener/timer が残らない
124+
- step deadline の意味（ACK後の worker budget）を維持する
125+
4) Map が増え続けない（`pendingRequests` / waiters 等）
126+
5) 新規ヘルパーは unit test で
127+
- timeout
128+
- interrupt
129+
- finalizer cleanup（全経路）
130+
を担保する
131+
132+
## 非目標
133+
134+
- コードベース全体を関数型に寄せる
135+
- Layer/Service による大規模DI（Effect Context）をいきなり全域に入れる
136+
- ログ/テレメトリを全面置換する
137+
138+
## リスクと軽減
139+
140+
1) 学習コスト・パラダイム混在
141+
- 最初は「内部ヘルパー層に閉じる」ことで拡散を防ぐ
142+
143+
2) 依存追加のフットプリント
144+
- 段階導入・限定的 import で影響を最小化
145+
146+
3) 意味のズレ（timeout と cancel の扱い）
147+
- 既存挙動を boundary でテストし、段階ごとに差分を小さくする
148+
149+
4) デバッグ/スタックトレース
150+
- 既存エラーメッセージを維持し、必要箇所だけトレースを活用する
151+
152+
## 参考
153+
154+
- Effect: https://github.com/Effect-TS/effect
155+
- 関連: `docs/issues/0005-supervised-ipc-submit-job-timeout.md`（IPC timeout/correlation）

‎docs/issues/0010-effect-adoption-structured-concurrency.md‎

Lines changed: 170 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,170 @@
1+
# Adopt Effect for structured concurrency (reduce Promise.race / AbortSignal / timer complexity)
2+
3+
Status: proposal (incremental adoption; no API-breaking rewrite)
4+
5+
## Problem
6+
7+
Roboppi currently implements cancellation / timeouts / concurrency using a mix of:
8+
9+
- `AbortController` / `AbortSignal` listeners
10+
- `setTimeout` / `clearTimeout`
11+
- `Promise.race` / ad-hoc backoff loops
12+
- manual cleanup in `finally` blocks
13+
14+
This is workable but has predictable failure modes as complexity grows:
15+
16+
1) duplicated patterns with subtle differences
17+
- `waitForAbort`, `sleep(ms, signal)`, `createScopedAbort`, and repeated `Promise.race([...])` exist in multiple places
18+
19+
2) cleanup correctness is hard to audit
20+
- listeners and timers must be removed on *every* exit path
21+
- request maps / waiter maps can leak when new early-return paths are added
22+
23+
3) error handling is fragmented
24+
- some paths intentionally suppress unhandled rejections (`p.catch(() => {})`)
25+
- some timeouts are modeled as exceptions, others as sentinel values (e.g. `null`)
26+
27+
4) concurrency semantics are implicit
28+
- cancellation propagation (parent abort -> child operations) is often implemented manually
29+
- multi-step flows (submit_job -> permit -> execute -> cancel) interleave timeouts and cancellation in ways that are hard to reason about
30+
31+
Representative hotspots:
32+
33+
- `src/workflow/core-ipc-step-runner.ts` (multiple races + scoped abort + cancel best-effort)
34+
- `src/ipc/protocol.ts` (`pendingRequests` + timers + fast-fail on disconnect)
35+
- `src/worker/process-manager.ts` (graceful shutdown uses multiple races / timeouts)
36+
- `src/scheduler/supervisor.ts` (spawn + transport bridging has several `new Promise` blocks)
37+
38+
## Goals
39+
40+
1) make cancellation / timeout behavior explicit and composable
41+
2) centralize cleanup via structured resource management
42+
3) reduce `Promise.race` / timer / listener boilerplate and the chance of leaks
43+
4) keep external behavior stable (Promise-returning public APIs can remain)
44+
5) preserve Bun + ESM runtime compatibility
45+
46+
## Proposed Approach (Effect-TS)
47+
48+
Adopt `effect` (Effect-TS) for *internal* concurrency primitives:
49+
50+
- interruption-aware async boundaries: `Effect.async((resume, signal) => ...)`
51+
- timeouts: `Effect.timeout` / `Effect.timeoutFail`
52+
- racing: `Effect.race`
53+
- structured finalization: `Effect.acquireRelease` (Scope)
54+
- retry/backoff: `Effect.retry` + `Schedule` (exponential + jitter)
55+
- structured concurrency: fibers (`Effect.fork`, `Fiber.interrupt`, `Effect.all/forEach` with concurrency)
56+
57+
Key idea: do not rewrite the entire codebase to be Effect-first. Instead:
58+
59+
- keep call sites `async/await`-friendly by bridging at module boundaries (`Effect.runPromise`)
60+
- move the most fragile Promise patterns behind small, well-tested Effect helpers
61+
62+
## Migration Plan (Staged)
63+
64+
### Phase 0: feasibility + footprint
65+
66+
- Add dependency: `bun add effect`
67+
- Validate Bun+ESM integration in CI (`make typecheck`, `make test`)
68+
- Decide a minimal internal wrapper surface (avoid importing many submodules everywhere)
69+
70+
Deliverable:
71+
72+
- new internal module (name TBD) that provides a tiny bridge layer
73+
- `runPromise` wrapper
74+
- `sleep`, `waitForAbort`, `withTimeout`, `raceAbort`, etc.
75+
76+
### Phase 1: IPC request/response waiter (highest leverage)
77+
78+
Target: `src/ipc/protocol.ts` `waitForResponse()`.
79+
80+
Replace the manual `{ timer, pendingRequests Map }` pattern with an Effect-based implementation:
81+
82+
- register waiter in a single place
83+
- guarantee cleanup in a finalizer (remove from map + clear timer)
84+
- fail-fast on transport close without callers needing to add special logic
85+
86+
Keep `waitForResponse()` returning `Promise<unknown>` for now:
87+
88+
- internal: Effect
89+
- boundary: `Effect.runPromise`
90+
91+
### Phase 2: Core IPC step flow (submit/permit/execute/cancel)
92+
93+
Target: `src/workflow/core-ipc-step-runner.ts`.
94+
95+
Goals:
96+
97+
- remove ad-hoc races (`Promise.race([completionPromise, waitForAbort(...)])`)
98+
- model cancellation and timeouts as interruption and structured timeouts
99+
- guarantee listener cleanup in one place
100+
101+
Nice-to-have:
102+
103+
- express "best-effort cancel then wait up to 5s for completion" as a clear combinator
104+
105+
### Phase 3: Process lifecycle (spawn + graceful shutdown)
106+
107+
Target: `src/worker/process-manager.ts`.
108+
109+
Replace multi-race shutdown logic with:
110+
111+
- `acquireRelease` for registration and abort-listener lifecycle
112+
- `timeoutFail` for grace windows
113+
- clear modeling of escalation: SIGTERM -> SIGKILL -> "stuck" fallback
114+
115+
### Phase 4: Supervisor spawn / transport bridging
116+
117+
Target: `src/scheduler/supervisor.ts`.
118+
119+
This file has several promise-wrapped callbacks and error/close paths.
120+
121+
- wrap callback-style writes / server close into `Effect.async`
122+
- make cleanup uniform (close server, rm temp dirs) via finalizers
123+
124+
### Phase 5 (Optional): Workflow-level structured concurrency
125+
126+
Target: `src/workflow/executor.ts`.
127+
128+
This is the biggest payoff but also the biggest rewrite.
129+
130+
- model step execution with fibers and a concurrency limit using `Effect.forEach(..., { concurrency })`
131+
- ensure workflow abort interrupts all child fibers
132+
- keep existing `ExecEventSink` emission semantics
133+
134+
## Acceptance Criteria
135+
136+
1) `make typecheck` passes
137+
2) `make test` passes
138+
3) no regressions in cancellation semantics:
139+
- parent abort cancels in-flight waits without leaking listeners/timers
140+
- step deadlines still apply to worker budget the same way as today
141+
4) no unbounded growth in internal maps (`pendingRequests`, waiters)
142+
5) new helpers have unit tests for:
143+
- timeout
144+
- interrupt
145+
- finalizer cleanup on all exit paths
146+
147+
## Non-goals
148+
149+
- converting the entire codebase to functional style
150+
- introducing Effect contexts (Layers/Services) across the whole runtime immediately
151+
- replacing existing logging/telemetry wholesale
152+
153+
## Risks & Mitigations
154+
155+
1) learning curve / mixed paradigm
156+
- mitigate by keeping Effect usage behind a small internal helper layer at first
157+
158+
2) new dependency footprint
159+
- mitigate by limiting surface imports and using staged adoption
160+
161+
3) subtle semantic drift (e.g., timeouts vs cancellation)
162+
- mitigate via tests that assert current behavior at module boundaries
163+
164+
4) stack traces / debugging ergonomics
165+
- mitigate by enabling Effect tracing only where needed and preserving existing error messages
166+
167+
## Reference
168+
169+
- Effect docs: https://github.com/Effect-TS/effect
170+
- Related: `docs/issues/0005-supervised-ipc-submit-job-timeout.md` (IPC timeouts / correlation)

‎package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
},
2323
"dependencies": {
2424
"@opentui/core": "0.1.80",
25+
"effect": "^3.19.19",
2526
"yaml": "^2.8.2"
2627
}
2728
}

0 commit comments

Comments
 (0)