Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 5 additions & 1 deletion plugins/ndf/scripts/hook_lib/ask_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,12 +105,16 @@ def run_request_ask(main: Path, plan: str, lane: str, payload: dict) -> tuple[in

def hold_text(rid: str, plan: str) -> str:
"""オーガナイザーへ送った問いの理由と保留の手順(正本: asking.md の「並列の計画のレーン」)。"""
import parallel_files

stop = parallel_files.stop_self() or f'python3 {parallel_files.RELAY} stop "$NDF_RELAY_NAME"'
loop = f'python3 "{Path(__file__).resolve().parent.parent / "sprint-state.py"}" loop hold-question <ループ> --ref {rid} --plan <止まったプラン> --step <続きのステップ> --parallel {plan}'
return "\n".join(
[
f"ndf: 並列の計画 {plan} のレーンの問いをオーガナイザーへ送った(依頼 {rid})。この作業を保留し、別の作業へ進め。答えは周回の境で届く。",
f"開発ループのレーン: {loop} を打ち、周回を質問待ちで締めて次の周回へ進む。",
f"1 スプリントのレーン・一時レーン: 引継ぎ文書の「次にやること」に続きのコマンドと依頼 {rid} を書き、ndf-next を出さずにセッションを終える(オーガナイザーが起こす)。",
f"1 スプリントのレーン・一時レーン: 引継ぎ文書の「次にやること」に続きのコマンドと依頼 {rid} を書き、ndf-next を出さずに、"
f"最後に {stop} を打って自分のラッパーを止めてセッションを終える(オーガナイザーが起こす)。",
f"正本: {ASKING_MD}",
]
)
Expand Down
14 changes: 14 additions & 0 deletions plugins/ndf/scripts/lib/parallel_files.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,25 @@
from __future__ import annotations

import json
import os
import shlex
from pathlib import Path
from typing import Optional

PARALLEL_DIR = Path(".ndf") / "parallel"
ROLE_FILE = "role.json" # 統括役の作業ディレクトリの役割のファイル(relay_lib.hub.ROLE_FILE と同じ公開の形。統括役を import しない)
RELAY = Path(__file__).resolve().parent.parent / "relay.py"


def stop_self(main: Optional[Path] = None, env=None) -> Optional[str]:
"""セッションの最後に自分のラッパーを止める 1 行(判断 fca6fe464619 の (a))。`ndf-next` を出さずに終えても対話の claude が
入力待ちで残り、ラッパーが止まらないためである。ラッパーの外(`NDF_RELAY_NAME` が無い)では None(打たない)。
`main` を渡すと、並列の計画のレーンか一時レーンのときだけ返す(利用者の端末のラッパーを止めない)。"""
env = os.environ if env is None else env
name = env.get("NDF_RELAY_NAME")
if not name or (main is not None and lane_name_in_plans(Path(main), name, env.get("NDF_RELAY_DIR")) is None):
return None
return f'python3 {shlex.quote(str(RELAY))} stop "$NDF_RELAY_NAME"'


def plans_in(main: Path) -> list[str]:
Expand Down
24 changes: 24 additions & 0 deletions plugins/ndf/scripts/lib/parallel_lane.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,30 @@ def lane_of(main: Path, lane: Optional[str] = None, status=run_status) -> Option
return None


def lane_waits(main: Path, name: str, lane: str, rid: str, relay_ls) -> bool:
"""答えの後に持ち主のレーンを起こすかを、レーンの側の待ちの記録で決める(6f)。ラッパーの状態では決めない(止め方の
食い違いで running が残っても起こし損ねないため)。ループのレーンは記録が `waiting_only` で止まっている、1 スプリントの
レーンは引継ぎ文書に依頼の id がある(保留の手順が書く)。記録を持たない一時レーンは、ラッパーが止まっているか
(`relay_ls()` の統一 JSON レポートで running / awaiting_approval でない)で決める(保留の手順の最後に自分で止める)。"""
from sprint_loop_names import LOOP_DIR

try:
hit = pp.load_valid_plan(main, name).lane(lane)
except (OSError, ValueError):
hit = None
if hit is None:
code, got = relay_ls()
state = next((i.get("state") for i in got.get("items") or [] if i.get("name") == lane), None) if code == 0 else None
return state not in ("running", "awaiting_approval")
try:
if isinstance(hit.input, pp.LoopInput):
rec = json.loads((Path(main) / LOOP_DIR / hit.input.loop / "loop.json").read_text(encoding="utf-8"))
return rec.get("status") == "stopped" and (rec.get("stop") or {}).get("condition") == "waiting_only"
return rid in (Path(main) / ".ndf" / "handoff" / f"{hit.handoff_name()}.md").read_text(encoding="utf-8")
except (OSError, ValueError, AttributeError):
return False


def lane_plans(main: Path, lane: Optional[str] = None) -> Optional[int]:
"""レーンなら、そのレーンの同時に流すプランの本数。レーンでなければ None(`queue --max` の既定)。"""
got = lane_of(main, lane)
Expand Down
7 changes: 5 additions & 2 deletions plugins/ndf/scripts/lib/parallel_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -486,8 +486,11 @@ def settle(self, rel: dict, msg: dict) -> None:
def helper_spawn(name: str, n: int, k: int, pending: Optional[str] = None) -> str:
"""一時レーンを起こす 1 行(`metrics.spawn`)。pending はループ名で、保留の周回の残りを流す一時レーン(異常系 11)。"""
what = f"pending {pending}" if pending else f"{n} {k}"
first = f"/goal /ndf:development-workflow --parallel-helper {name} {what} -- 終わり: [ndf-parallel] helper-done {name} の行を出したとき"
return f"python3 {RELAY} spawn --name {pr.helper_name(name, n)} --role helper --plan {name} --json -- '{first}'"
helper = pr.helper_name(name, n)
# 終わりに自分のラッパーを止める(判断 fca6fe464619 の (a))。止めないと対話の claude が入力待ちで残り、続きの一時レーンが断られる
end = f"[ndf-parallel] helper-done {name} の行を出し、python3 {RELAY} stop {helper} で自分のラッパーを止めたとき"
first = f"/goal /ndf:development-workflow --parallel-helper {name} {what} -- 終わり: {end}"
return f"python3 {RELAY} spawn --name {helper} --role helper --plan {name} --json -- '{first}'"


def present_approval(d: Path) -> str:
Expand Down
11 changes: 4 additions & 7 deletions plugins/ndf/scripts/lib/sprint_loop_record.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import handoff_doc
import jsonio
import locks
import parallel_files
import project_mvv as pm
import repo as repo_lib
import step_result
Expand Down Expand Up @@ -382,13 +383,9 @@ def stopped_result(env: Env, rec: dict, summary: str, extra_metrics: dict | None
resume = resume_command(rec)
update_handoff(env, rec, f"止まった({stop['condition']})。理由が解けたら再開のコマンドで続ける", resume, resume)
items = [{"condition": stop["condition"], "values": stop["values"], "resume": resume}]
metrics = {
"status": "stopped",
"stop": stop,
"stop_line": stop_line(rec, stop["condition"]),
"consumed": consumed(rec),
**(extra_metrics or {}),
}
metrics = {"status": "stopped", "stop": stop, "stop_line": stop_line(rec, stop["condition"]), "consumed": consumed(rec)}
# 待ちだけで止まったレーンは、自分のラッパーを止める 1 行(stop_self)を打って終える(判断 fca6fe464619 の (a))
metrics.update(extra_metrics or {}, stop_self=parallel_files.stop_self(env.main_dir()) if stop["condition"] == "waiting_only" else None)
return {**step_result.result(TOOL, "stopped", summary, items, metrics), "exit": EXIT_STOPPED}


Expand Down
8 changes: 4 additions & 4 deletions plugins/ndf/scripts/parallel.py
Original file line number Diff line number Diff line change
Expand Up @@ -365,17 +365,17 @@ def cmd_request_gate2(a) -> None:


def _owner_restart(main: Path, name: str, rid: str, choice: list[str] | None = None) -> str | None:
"""答えの後の 1 行。自分で立てた問い(stall・helper・spawn)は選んだ答えの commands、ほかは止まった持ち主の restart。"""
"""答えの後の 1 行。自分で立てた問い(stall・helper・spawn)は選んだ答えの commands、ほかは答えを待って止まった持ち主の
restart。起こす要否はレーンの側の待ちの記録で決める(`parallel_lane.lane_waits`)。"""
import durable
import parallel_run as pr_run

items = (durable.event_from(pl.KIND, pl.record_identity(main, name), pl.workflow_id(name), "questions", 0) or {}).get("items") or []
q = next((q for q in items if q.get("id") == rid), {})
if q.get("kind") in ("stall", "helper", "spawn") or not (lane := q.get("lane")):
return next((q["commands"][c] for c in choice or [] if (q.get("commands") or {}).get(c)), None)
code, got = pr_run.relay(main, "ls", "--json")
state = next((i.get("state") for i in got.get("items") or [] if i.get("name") == lane), None) if code == 0 else None
return None if state in ("running", "awaiting_approval") else f"python3 {pr_run.RELAY} restart {lane}"
waits = pl.lane_waits(main, name, lane, rid, lambda: pr_run.relay(main, "ls", "--json"))
return f"python3 {pr_run.RELAY} restart {lane}" if waits else None


def cmd_answer(a) -> None:
Expand Down
10 changes: 10 additions & 0 deletions plugins/ndf/scripts/tests/test_ask_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -353,6 +353,16 @@ def test_a_lane_question_is_sent_to_the_organizer_and_denied(proj, monkeypatch,
assert (plan, lane) == ("p1", "la") and sent["tool_input"]["metadata"]["source"] == "ndf-ask:cost" and sent["ledger_id"]


def test_the_hold_steps_end_by_stopping_the_own_wrapper(proj, monkeypatch, forwarded):
"""保留の手順は自分のラッパーを止めて終える(判断 fca6fe464619 の (a))。ndf-next を出さずに終えても対話の claude が残るためである。"""
import parallel_files

lane_plan(proj)
monkeypatch.setenv("NDF_RELAY_NAME", "la")
out = json.dumps(decide(proj, "ndf-ask:cost"), ensure_ascii=False)
assert parallel_files.stop_self() and json.dumps(parallel_files.stop_self(), ensure_ascii=False)[1:-1] in out


def test_a_helper_question_is_sent_by_the_role_file(proj, monkeypatch, forwarded, tmp_path):
lane_plan(proj)
relay = tmp_path / "relay"
Expand Down
48 changes: 48 additions & 0 deletions plugins/ndf/scripts/tests/test_parallel_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -664,3 +664,51 @@ def test_answer_to_an_own_question_returns_the_chosen_command(monkeypatch, tmp_p
monkeypatch.setattr(durable, "event_from", lambda *_a, **_k: {"items": [item]})
assert cli._owner_restart(tmp_path, "p1", "h1", ["続きから起こし直す"]) == "SPAWN"
assert cli._owner_restart(tmp_path, "p1", "h1", ["止める"]) is None


def test_owner_is_woken_by_the_lane_side_wait_record(monkeypatch, tmp_path):
"""持ち主を起こす要否はラッパーの状態でなくレーンの側の待ちの記録で決める。ループは記録が waiting_only で止まっている、
1 スプリントのレーンは引継ぎ文書に依頼の id がある。"""
import importlib.util

spec = importlib.util.spec_from_file_location("parallel_cli", Path(prun.__file__).resolve().parents[1] / "parallel.py")
cli = importlib.util.module_from_spec(spec)
spec.loader.exec_module(cli)
data = {
"version": 1,
"name": "p1",
"rules": [],
"lanes": [
{"name": "la", "input": {"kind": "loop", "loop": "l1", "args": ["--milestones", "01 a"]}, "plans": 1, "touches": ["a/"]},
{"name": "lb", "input": {"kind": "sprint", "issues": [7]}, "plans": 1, "touches": ["b/"]},
],
}
pp.plan_dir(tmp_path, "p1").mkdir(parents=True)
(pp.plan_dir(tmp_path, "p1") / "plan.json").write_text(json.dumps(data))
items = [{"id": "qa", "kind": "ask", "lane": "la"}, {"id": "qb", "kind": "ask", "lane": "lb"}]
monkeypatch.setattr(durable, "event_from", lambda *_a, **_k: {"items": items})
monkeypatch.setattr(
prun, "relay", lambda *_a, **_k: (0, {"items": [{"name": "la", "state": "running"}, {"name": "lb", "state": "running"}]})
)
loop = tmp_path / ".ndf" / "loop" / "l1"
loop.mkdir(parents=True)
handoff = tmp_path / ".ndf" / "handoff"
handoff.mkdir(parents=True)
assert cli._owner_restart(tmp_path, "p1", "qa") is None and cli._owner_restart(tmp_path, "p1", "qb") is None
(loop / "loop.json").write_text(json.dumps({"status": "running"}))
(handoff / "issue-7.md").write_text("## 次にやること\n- 別の作業\n")
assert cli._owner_restart(tmp_path, "p1", "qa") is None and cli._owner_restart(tmp_path, "p1", "qb") is None
(loop / "loop.json").write_text(json.dumps({"status": "stopped", "stop": {"condition": "waiting_only"}}))
(handoff / "issue-7.md").write_text("## 次にやること\n- 依頼 qb の答えを読んで続ける\n")
assert cli._owner_restart(tmp_path, "p1", "qa") == f"python3 {prun.RELAY} restart la"
assert cli._owner_restart(tmp_path, "p1", "qb") == f"python3 {prun.RELAY} restart lb"
items.append({"id": "qh", "kind": "ask", "lane": "p1-rel1"}) # 記録を持たない一時レーンはラッパーが止まっているかで決める
assert cli._owner_restart(tmp_path, "p1", "qh") == f"python3 {prun.RELAY} restart p1-rel1"


def test_a_helper_ends_by_stopping_its_own_wrapper():
"""一時レーンは helper-done の後に自分のラッパーを止めて終える(判断 fca6fe464619 の (a))。止めないと対話の claude が
入力待ちで残り、続きの一時レーンの起動が同じ名前か上限で断られる。配布と保留の周回のどちらの起動も同じ。"""
for line in (prun.helper_spawn("p1", 2, 1), prun.helper_spawn("p1", 3, 0, pending="lroot")):
name = line.split("--name ", 1)[1].split()[0]
assert f"{prun.RELAY} stop {name}" in line.split(" -- ", 1)[1]
26 changes: 26 additions & 0 deletions plugins/ndf/scripts/tests/test_sprint_loop.py
Original file line number Diff line number Diff line change
Expand Up @@ -1386,6 +1386,32 @@ def test_waiting_only_instead_of_no_issue_when_only_pending_work_is_left(env):
assert out["exit"] == 3 and out["metrics"]["stop"]["condition"] == "waiting_only"


def test_a_waiting_only_lane_stops_its_own_wrapper(env, monkeypatch):
"""待ちだけで止まったレーンは自分のラッパーを止める 1 行を返す(判断 fca6fe464619 の (a))。ラッパーの外と、並列の計画の
レーンでないラッパーでは返さない。"""
import parallel_files

request_lap(env)
env.rank_box["out"] = rank_out(ms("01 近い", [11, 12]))
monkeypatch.delenv("NDF_RELAY_NAME", raising=False)
out = sl_select.cmd_select(env, "l1")
assert out["metrics"]["stop"]["condition"] == "waiting_only" and out["metrics"].get("stop_self") is None
sl.cmd_start(env, "l1", [], {})
monkeypatch.setenv("NDF_RELAY_NAME", "la")
out = sl_select.cmd_select(env, "l1") # 並列の計画のレーンでないラッパー(利用者の端末)は止めない
assert out["metrics"]["stop"]["condition"] == "waiting_only" and out["metrics"].get("stop_self") is None
plan = env.root / ".ndf" / "parallel" / "p1"
plan.mkdir(parents=True)
(plan / "plan.json").write_text(json.dumps({"name": "p1", "lanes": [{"name": "la"}]}))
sl.cmd_start(env, "l1", [], {})
out = sl_select.cmd_select(env, "l1")
assert out["metrics"]["stop"]["condition"] == "waiting_only" and out["metrics"]["stop_self"] == parallel_files.stop_self(env.root)
assert out["metrics"]["stop_self"]
sl.cmd_start(env, "l1", [], {})
sl.cmd_stop(env, "l1")
assert sl_select.cmd_select(env, "l1")["metrics"].get("stop_self") is None # 待ちでない止まり方はラッパーを残す


def test_budget_waits_first_but_user_stop_stops_at_once(env):
request_lap(env)
r = rec(env)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ hook が `parallel.py request-ask` でオーガナイザーへ送り、問いを
| 送り手 | 保留の仕方 | 答えが届いたとき |
| --- | --- | --- |
| 開発ループのレーン | `sprint-state.py loop hold-question <ループ> --ref <依頼> --plan <止まったプラン> --step <続きのステップ>` で周回を「質問待ち」で締め、次の周回へ進む([loop.md](loop.md)) | 周回の境で、止まったプランを続きのステップから流す |
| 1 スプリントのレーン・一時レーン | 引継ぎ文書の「次にやること」に続きのコマンドと依頼の id を書き、`ndf-next` を出さずにセッションを終える | オーガナイザーが `relay.py restart <レーン>` で起こす。`parallel.py await --from` か答えのイベントを読み、続きから進む |
| 1 スプリントのレーン・一時レーン | 引継ぎ文書の「次にやること」に続きのコマンドと依頼の id を書き、`ndf-next` を出さずに、最後に `relay.py stop "$NDF_RELAY_NAME"` で自分のラッパーを止めてセッションを終える(1 行は保留の手順が返す) | オーガナイザーが `relay.py restart <レーン>` で起こす(起こす要否は引継ぎ文書の依頼の id で決める)。`parallel.py await --from` か答えのイベントを読み、続きから進む |

設計の承認(承認ゲート 1)と本番の配布(承認ゲート 2)は、レーンが依頼として送るため問わない。ループのレーンは承認ゲート 1 で
課題を人の判断待ちへ移すときに設計の承認の依頼を送り、本番のステージは配布の依頼のプランになる。
Expand Down
4 changes: 4 additions & 0 deletions plugins/ndf/skills/development-workflow/references/loop.md
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,10 @@
(`[ndf-loop] stopped <名> <condition>`)をそのまま出して**応答を終える。再開のコマンドは引継ぎ文書 `loop-<名>` の
「次に実行するコマンド」にもスクリプトが書く。

`waiting_only` で止まった並列の計画のレーンは、結果に `metrics.stop_self`(自分のラッパーを止める 1 行)が付く。止まった行を出した後、
最後にその 1 行を打って終える。打たないと対話の claude が入力待ちで残り、ラッパーが止まらない。オーガナイザーは、ループの記録が
`waiting_only` で止まっていることを見て、答えの後に `relay.py restart` で起こす。

| 止まった理由 | 利用者へ問うか |
| --- | --- |
| `gate2_returned` | 問う(承認ゲート 2) |
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -74,8 +74,8 @@
最初の入力は次のどちらかで、`development-workflow` は判定の手順へ進まずこの節の手順で進める。

```text
/goal /ndf:development-workflow --parallel-helper p1 <連番> <提示> -- 終わり: [ndf-parallel] helper-done p1 の行を出したとき
/goal /ndf:development-workflow --parallel-helper p1 pending <ループ名> -- 終わり: [ndf-parallel] helper-done p1 の行を出したとき
/goal /ndf:development-workflow --parallel-helper p1 <連番> <提示> -- 終わり: [ndf-parallel] helper-done p1 の行を出し、python3 <relay.py> stop p1-rel<連番> で自分のラッパーを止めたとき
/goal /ndf:development-workflow --parallel-helper p1 pending <ループ名> -- 終わり: [ndf-parallel] helper-done p1 の行を出し、python3 <relay.py> stop p1-rel<連番> で自分のラッパーを止めたとき
```

**配布(`<連番> <提示>`):**
Expand All @@ -87,7 +87,9 @@
- `phase: confirmed`: 答えの記録 → 提示のプランを `approve` から → 本番のプランの queue と待ち。`approve` が承認ゲート(終了コード 10)で
終われば本番のプランを流さず、問いとして送って手順 4 へ進む
3. `metrics.settle` の 1 行(`parallel.py settle p1 <連番> --done <D>`)を打つ。結果は耐久ワークフローが範囲のレーンへ返す
4. `[ndf-parallel] helper-done p1` の行を出して終える。答えの後は、オーガナイザーが続きの一時レーンを起こす
4. `[ndf-parallel] helper-done p1` の行を出し、最後に最初の入力の「終わり」にある `relay.py stop <自分の名前>` の 1 行を打って
自分のラッパーを止めて終える。止めないと対話の claude が入力待ちで残り、続きの一時レーンの起動が同じ名前か上限で断られる。
答えの後は、オーガナイザーが続きの一時レーンを起こす

CI の失敗・衝突・版数の衝突は、配布のプランの直しの手順(judge・fix)で直す。直せない失敗は問いにする(区分つきの `AskUserQuestion` を
出すと、問いの hook がオーガナイザーへ送り、拒否の理由に保留の手順が返る)。保留したら手順 4 で終える。
Expand Down
Loading