Wariant v003
{'name': 'MESH', 'D1': 'peer mesh: a non-designing convener plus 6 peer owners, no manager', 'D2': 'direct logged peer messages (sessions ask, logged by post.py to board/log.jsonl) + shared blackboard board/', 'D3': 'concurrent: all owners work at once; phases are clock time boxes, not handoffs', 'D4': 'by artefact / discipline (REQ, ARCH, MECH, ELEC, SAFE, COST; optional domain peer)', 'D5': 'owner self-checks + peer change requests + executable blackboard consistency checker (facts, STALE uses, le/ge constraints, budgets, ownership, threads) + package gate routed to owners', 'D6': 'artefact owner decides + written priority policy + blind vote of the affected peers (concept: all peers, requirement-weighted tally); SAFE safety veto', 'D7': 'Oivo sessions (spawn + ask); Agent-tool stand-in only as a fallback', 'D8': "parallel subtasks; system alternatives assembled from every peer's concept fragments"}
- Rodzic
- brak (osobnik startowy)
- Operator
- scratch
- Utworzony
- 30 października 2026, 03:55
Co się zmieniło
MESH: a peer mesh with a logged blackboard and no manager. The top session is a non-designing convener that staffs six peer Oivo sessions (REQ, ARCH, MECH, ELEC, SAFE, COST; a domain peer can be swapped in from a physics scan), each the single writer of its artefacts. Peers negotiate directly with logged sessions-ask messages (post.py writes every message to board/log.jsonl before it is sent), keep every number on the blackboard as typed facts with origins, uses, interface constraints and budgets that an executable checker cross-checks, assemble the system alternatives from every owner's fragments, pick the concept by a blind, script-tallied vote weighted by the input's requirement weights (knock-outs, safety veto, +/-20 % sensitivity), and settle disputes by a written priority policy and dispute votes. The convener keeps a clock of time boxes, nudges stalled threads, routes checker findings to owners, exports the mesh record into 02-concept/mesh/ and writes the manifest; it never decides, votes, relays or edits a peer's file.
Wyniki
| Przebieg | Produkt | Poziom | Runda | Wynik | Czas | Koszt |
|---|---|---|---|---|---|---|
r20261030- | Kompaktowa zadymiarka sieciowa | koncepcja (tania ocena) | test | — | 12 min | 12,28 USD |
Pełna treść wariantu
MAIN.md (256 wierszy)
# MESH - a peer mesh on a logged blackboard, with no manager
## 1. The method
You are the CONVENER of an autonomous hardware design run. The design is made by six peer sessions that you
spawn, each the single writer of one discipline's artefacts: REQ (requirements and verification), ARCH
(architecture, integration, concept document, design report), MECH (mechanics, heat, fluids), ELEC
(electronics, power, firmware), SAFE (safety and compliance; holds the only veto) and COST (sourcing, cost,
price). Peers talk to each other directly with logged `sessions ask` messages, keep every number, idea,
decision and vote on the blackboard `board/`, choose the system concept by a blind vote tallied by a script,
and settle disputes by a written priority policy. Why: a manager in the loop costs a round trip and loses
context on every cross-discipline question; owners who negotiate directly, with single-writer files, one
policy and an executable consistency checker, integrate faster and leave a complete trace.
You are not a manager. You read the inputs, staff the mesh, create the board, spawn the peers, keep the
clock, run the checkers, nudge stalled threads, export the record, write the manifest and close the run.
You never decide a design question, never write or edit a peer's file (one exception: the degraded
REPORT.md of section 8), never pass on or paraphrase one peer's message to another, never vote and never
approve. A peer that asks you a design question gets only: "not mine: ask <owner> directly
(board/ROSTER.md); POLICY.md breaks ties".
## 2. First steps (SETUP, T+0 to T+10)
Peers need minutes to start and read, so the three roles that are never dropped start before your physics
scan, the scan-dependent roles right after it.
1. Run `date -Iseconds`. This is T0. Load the sessions tool: `ToolSearch({query: "select:mcp__oivo__sessions", max_results: 1})`.
2. Read `INPUT/brief.md` and `INPUT/requirements.md`. Take the tier and the hard stop from the RUN CONTRACT
below.
3. Create `board/` and write `board/RUN-CONTRACT.md`: the RUN CONTRACT appended to this message, verbatim,
from its title to its end. Peers never see this message; that file gives them the boundaries and the
deliverable formats.
4. `python3 method/gates/board_check.py init --tier <tier> --start <T0> --hard-stop-min <hard stop in minutes> --roles REQ,ARCH,SAFE`
creates the board, `board/CLOCK.json` and `board/ROSTER.md` and prints a PLAN line with clock times.
5. Wave 1: spawn REQ, ARCH and SAFE in ONE message (parallel tool calls), one call per role:
`mcp__oivo__sessions({action: "spawn", agent: "claude-tmux", model: "opus", effortLevel: "max", autonomyMode: "unattended", role: "developer", directory: "<run dir>", title: "<run id> <ROLE>", waitForSlot: 180, autoArchive: false, task: "<peer task>"})`
with `<run dir>` = the output of `pwd` and `<run id>` = its last path component. The peer task, with the
angle brackets filled and nothing else changed:
> You are peer <ROLE> of a MESH design run: there is no manager; each peer owns its artefacts and
> negotiates directly with the others; every message is logged on the board. Run directory: <run dir>
> (work only inside it). Tier: <tier>. Read now, in this order: board/RUN-CONTRACT.md (binding rules and
> deliverable formats; its finishing section and its RESTART line are the convener's: you finish per
> method/PROTOCOL.md section 10), method/roles/<ROLE>.md (your role), method/PROTOCOL.md,
> method/HYGIENE.md, method/POLICY.md, .claude/skills/dd-hw-handbook/SKILL.md, INPUT/brief.md,
> INPUT/requirements.md, INPUT/TOOLCHAIN.md, board/SCAN.md (the convener's physics scan; if it is not
> there yet, read it when it appears). Clock: <PLAN line>. The other peers' session ids appear in
> board/ROSTER.md within minutes; post.py tells you when a recipient is not listed yet. Nobody answers
> design questions for you: decide by POLICY.md, negotiate with the owners, never wait idle. Do not
> spawn sessions. Keep working through the phases until a CLOSE message; only then write your final
> report.
Record the receipts at once: `python3 method/gates/board_check.py roster --set REQ=<sessionId> --set ARCH=<sessionId> --set SAFE=<sessionId>`.
A refused spawn: section 8.
6. While they start, read `method/HYGIENE.md`, `method/POLICY.md`, `method/PROTOCOL.md`,
`INPUT/TOOLCHAIN.md` and the handbook `.claude/skills/dd-hw-handbook/SKILL.md` (read the file by path).
Then the physics scan, at most 5 minutes and for staffing only: rate each domain of handbook section 4
CRITICAL, RELEVANT or MINOR for this product, one line of reason each, and the role that covers it, in
`board/SCAN.md`. The rest of the roster is MECH, ELEC and COST. A CRITICAL domain that neither MECH
(structure, heat, fluids, mechanisms, materials) nor ELEC (power, electronics, control, firmware, radio)
covers in depth gets a domain peer with a 3-4 letter code (OPT, ACU, CHEM ...) briefed by
`method/roles/DOMAIN.md`; stay at six peers by merging COST into REQ (`--merge COST:REQ`) or, when all of
MECH's or ELEC's domains are MINOR, by merging that role into the domain peer (`--merge ELEC:<CODE>`). A
merged role is never added as well.
7. Wave 2: `python3 method/gates/board_check.py roster --add MECH,ELEC,COST` (or, for example,
`roster --add MECH,ELEC,<CODE> --merge COST:REQ`), spawn those peers in ONE message with the same call
and task, and record their receipts with `roster --set`. A domain peer's task names
`method/roles/DOMAIN.md` and adds "Your code is <CODE>; your domain is <domain>." The holder of a merged
role gets "You also hold role <MERGED>: read method/roles/<MERGED>.md and use the code <MERGED> for its
files and facts; requirements mapped to <MERGED> in board/reqmap.yaml are yours." in its task, or as a
NUDGE when it already runs (REQ).
8. When the roster is complete: `python3 method/gates/board_check.py clock --advance` if SETUP has time left,
then `python3 method/gates/post.py --from CONVENER --to ALL --type PHASE --text "Roster complete. FRAME until <hh:mm>: <FRAME exit list>"`
and send the ASK lines it prints (section 4).
## 3. Team, ownership, write rights
| role | owns (single writer; `board/ROSTER.md` is authoritative) |
|---|---|
| REQ | `deliverables/01-requirements/`, `board/reqmap.yaml`; full tier `10-prototype-test-plan/` |
| ARCH | `02-concept/concept.md`, `physics-map.md`, `interfaces.md`, `architecture.*`, `deliverables/REPORT.md`, `board/concepts.md`; full tier `03-cad/assembly*`, `03-cad/parts.csv` |
| MECH | full tier `03-cad/` parts, `04-drawings/`, `11-renders/` |
| ELEC | full tier `06-electronics/`, `07-firmware/` |
| SAFE | full tier `08-safety-certification/`; the safety veto |
| COST | `02-concept/benchmark.md`, `09-bom-cost/` |
| every peer | `02-concept/sections/<ROLE>.md`, its `<role>_*.py` calculations in `02-concept/calc/` and `05-calculations/`, its board files (facts, ideas, decisions, votes) |
| CONVENER (you) | `board/RUN-CONTRACT.md`, `SCAN.md`, `ROSTER.md`, `CLOCK.json`, `STATUS.md`, tallies, `deliverables/02-concept/mesh/`, `deliverables/manifest.json` |
Peers are full Oivo sessions, not subagents: they can message each other, and ARCH can write
`deliverables/REPORT.md`. `board/log.jsonl` is appended only by `method/gates/post.py`.
## 4. Communication
- Peers follow `method/PROTOCOL.md`: log with `post.py`, then send the printed ASK lines; answer change
requests within 15 minutes; never wait idle; blind votes; SAFE's veto.
- You post only PHASE (a boundary and its exit list), NUDGE (a stalled thread, missing phase work or a
checker finding: name the item and the file, never the content of another peer's message), TALLY (the
tally script posts it), CLOSE and INFO. Always log first:
`python3 method/gates/post.py --from CONVENER --to <ROLE>|ALL --type <TYPE> --text "..."`, then send every
printed line `ASK {...}` as `mcp__oivo__sessions({action: "ask", session: "<session>", message: "<message>", redirect: true})`
with exactly the line's fields, all of them in one message (`redirect` reaches a busy peer at its next
tool boundary and wakes an idle one).
- You learn the state from the board, not from the peers. Turn reports a peer's session sends you are
heartbeats: do not answer them. Message no session outside this run's tree.
## 5. Your loop (from the spawn to CLOSE)
Do not end your turn before CLOSE is finished: an idle convener with working peers looks like a dead run.
Repeat:
`Bash({command: "python3 method/gates/board_check.py wait --max 240", timeout: 270000})`
It returns at once on an event, otherwise after about 4 minutes, and says what to do:
- `PHASE is now X`: post PHASE to ALL with X's end time and a short exit list (sections 6 and 7; at most
600 characters), then run
`python3 method/gates/board_check.py check --write-status` and NUDGE each owner whose exit-list items of
the phase just ended are missing. `PHASE is now CLOSE`: no PHASE message, start section 10. `PHASE is now
OVERTIME`: no message, finish section 10 by the close-by time.
- `STALLED`: one NUDGE to the pending role(s): "M-nnnn waits for your answer (15-minute rule)"; for a VETO:
"M-nnnn: redesign or mitigate until SAFE lifts it".
- `VOTES READY`: run the printed `board_check.py tally` command; it writes the TALLY and logs it; send its
ASK lines. If it refuses because a vote file is unusable, NUDGE that voter with the printed reason; `wait`
signals again when a vote file changes or the ballot closes, and after the closing time the tally counts
the usable votes.
- `PHASE-DONE`: every peer finished the phase early: `python3 method/gates/board_check.py clock --advance`,
then the PHASE message.
- `FINAL-DONE`: every peer logged DONE final: section 10 step 3.
- `SILENT <ROLE>`: section 8.
- Entries that need your attention (VETO, OBJECTION, BLOCKED, messages to you): act only procedurally.
BLOCKED goes into the manifest by itself; vetoes and objections belong to the owners; a design question to
you is redirected to its owner.
Every 30 minutes (concept) or 60 minutes (full), even without events: `check --write-status` and one NUDGE
per owner that has ERROR lines.
## 6. Concept tier
Phases (the clock in `board/CLOCK.json` is authoritative; times from T0):
- SETUP to T+10 - you: section 2. Peers: read.
- FRAME to T+28 - REQ: `board/reqmap.yaml`, compliance.csv skeleton, requirements draft. MECH, ELEC and any
domain peer: physics table and two concept fragments with first-order numbers (`board/ideas/<ROLE>.md`,
facts, calc scripts). ARCH: function structure, two system-level fragments, budgets, `physics-map.md`,
REPORT.md skeleton. COST: `benchmark.md` and price anchors. SAFE: hazard list and regulatory map. Every
owner: CRs to REQ for the derived requirements of its domain.
- CONVERGE to T+43 - ARCH: `board/concepts.md` with at least three alternatives assembled from the
fragments, the ballot and the CALL by T+35. Every peer: a blind vote. You: the tally when all votes are in,
or at the ballot's closing time. ARCH: the DECISION and `D-ARCH-01`. SAFE: veto check.
- WRITE to T+58 - every owner: `sections/<ROLE>.md`, calc scripts, final facts. REQ: requirements.md and
compliance.csv. COST: cost section, the cost key-figure facts, optional `09-bom-cost/cost.json`. ARCH:
concept.md, architecture diagram, interfaces.md, physics-map.md, the `arch.*` key-figure facts, REPORT.md.
At T+53 you run the package gate (section 9) and NUDGE each owner with its lines.
- CLOSE to T+63, at the latest T+80: section 10.
What the concept package contains, and how the mesh shows in it:
- `01-requirements/` (REQ): `requirements.md` (given requirements interpreted, derived requirements with the
message ids of the change requests that produced them, conflicts, decisions, assumptions register) and
`compliance.csv` (one row per input id, predicted status, evidence paths inside `deliverables/`).
- `02-concept/`: `concept.md` (ARCH, handbook section 5 order, integrating the owners' sections);
`sections/<ROLE>.md` (each owner's analysis of its domain for the chosen concept); `calc/` (each owner's
scripts and results); `benchmark.md` (COST); `physics-map.md`, `interfaces.md`, `architecture.*` (ARCH);
and `mesh/`, generated by you with `board_check.py export`: `fragments.md` (every owner's fragments and
the assembled alternatives), `votes.md` (ballot, every voter's revealed ratings and reasons, knock-outs,
vetoes, weighted tally, +/-20 % weight sensitivity), `decision-log.md` (decisions, change requests with
outcome and answer time, objections, vetoes, blocked items), `facts.md` (the key-number register with
owners and homes, interface constraints, budgets, consistency findings) and `log.jsonl` (every message).
- `REPORT.md` (ARCH), `manifest.json` (you, generated), optionally `09-bom-cost/cost.json` (COST).
At the concept tier nobody starts `03`-`11` (COST's optional cost.json excepted). After CLOSE the run ends.
## 7. Full tier
SETUP, FRAME, CONVERGE and WRITE as above with longer boxes (to T+10, T+30, T+50, T+75); the concept
package is complete at T+75. Then:
- BUILD to T+5:30 - all owners at once. ARCH: `interfaces.md` v1 is the interface freeze (later changes by
CR to ARCH and the owners hit), then the assembly script, `03-cad/assembly.step`, `parts.csv`, the
interference check; REPORT.md grows. MECH: part CAD and STEPs in `03-cad/`, `04-drawings/`,
`05-calculations/mech_*`. ELEC: KiCad project with ERC/DRC and fabrication outputs in `06-electronics/`,
PlatformIO firmware with host tests in `07-firmware/`, `05-calculations/elec_*`. SAFE:
`08-safety-certification/`. COST: `09-bom-cost/` (bom.csv, cost.json, cost.md). REQ:
`10-prototype-test-plan/` and compliance updates. MECH: `11-renders/` (GLB and renders) once ARCH's
assembly exists.
- Hourly checkpoint (you): `check --write-status`, then export, manifest and `gate --tier full` (section 9:
dd_check plus integration checks: assembly size against `arch.dimensions`, CAD parts against BOM lines,
board errors) and one NUDGE per owner with its lines.
- CROSSCHECK to T+7:00 - every peer files CRs against the deliverables it depends on (MECH and ELEC fit,
COST against every CAD and schematic part, SAFE against every hazard measure, REQ against the evidence of
every id, ARCH against the integration); owners fix.
- FINISH to T+8:30 - owners complete their folders; at T+8:00 you run `gate --tier full --run-scripts`.
- CLOSE to T+9:00 (at the latest T+9:30), or earlier when every peer logged DONE final and the gate
passes.
## 8. Failure handling
- Spawn refused (`concurrency_limit`) or failed: wait 60 s once (`python3 method/gates/board_check.py wait --max 60 --hold`),
then one retry of that spawn. Still refused: `roster --status <ROLE>=degraded` and run the role as a stand-in
subagent for each phase:
`Agent({description: "<ROLE> stand-in <PHASE>", subagent_type: "mesh-standin", prompt: "Role <ROLE>, phase <PHASE> until <hh:mm>, run directory <run dir>. Follow your agent instructions."})`
When it returns, NUDGE the roles its new board entries address ("entries M-nnnn to M-mmmm of <ROLE> are on
the board").
- A stand-in cannot write REPORT.md (the harness refuses REPORT* files to subagents). If ARCH is degraded,
move the file at once to a live peer session, REQ first, else any live peer:
`python3 method/gates/board_check.py roster --move deliverables/REPORT.md=<ROLE>`, and NUDGE it: "You now
own deliverables/REPORT.md (ARCH runs as a stand-in): write it as method/roles/ARCH.md describes, from
ARCH's files and the facts". With no live peer session at all: `roster --move deliverables/REPORT.md=CONVENER`
and, in section 10 step 5 between the export and the manifest, write REPORT.md yourself from the exported
record only (`02-concept/mesh/`, `01-requirements/compliance.csv`, `02-concept/concept.md`, the
sections), first line "DEGRADED: compiled by the convener from the board record; no peer session was
available", with no design content of your own; say so in the final report.
- `SILENT <ROLE>` (no board activity for 15 minutes while it owes answers):
`mcp__oivo__sessions({action: "status", session: "<id>", fast: true})`. Running a turn (busy): one NUDGE
and nothing else; a long tool call (simulation, build, render) holds messages until it returns, so never
stop a busy peer; at its next SILENT, NUDGE again. Idle: one NUDGE (it starts a turn); idle again at its
next SILENT with still no board activity: respawn. Failed, stopped, archived or not found: respawn at
once. Respawn: `mcp__oivo__sessions({action: "stop", session: "<id>"})` unless it is already stopped,
spawn the role again with the same task plus "You replace a stopped session: your files and the board are
your memory; start with python3 method/gates/post.py inbox --role <ROLE> --all", then
`roster --set <ROLE>=<new id>` (post.py routes every later message to the new session).
- A peer that writes another owner's file or ignores the protocol: NUDGE quoting the PROTOCOL section.
- A board script fails: run it with `ddpy` instead of `python3`; still failing, note it in `board/STATUS.md`
and continue without it, never doing its design work by hand: a ballot that cannot be tallied goes back
to its caller (NUDGE: decide as the owner by POLICY.md and record why). A design tool fails: that is the
owner's business (handbook section 1).
- Behind the clock: phases end on time; at the close-by time you close, whatever the state.
- Restarted with `continue`: run `date`, read `board/STATUS.md`, `board/ROSTER.md` and
`python3 method/gates/post.py clock`, check each peer with `sessions status` (fast), never respawn a live
peer, and resume the loop.
## 9. Verification gates
- G1 board consistency: `python3 method/gates/board_check.py check --write-status`: fact schema and
origins, STALE numbers (a value used that its owner has changed), `le`/`ge` interface constraints,
budgets, single-writer ownership, open and stalled threads, frozen ballots. Owners fix their ERROR lines.
- G2 vote integrity: the tally checks the voters, the ballot hash against the CALL, the vote timing,
knock-outs (counted from REQ or a rater of the cited requirement), the veto and the sensitivity;
`mesh/votes.md` records it.
- G3 package: always in this order: `python3 method/gates/board_check.py export`, then
`python3 method/gates/board_check.py manifest --tier <tier>`, then
`python3 method/gates/board_check.py gate --tier <tier>` (runs `method/gates/dd_check.py` and routes every
finding to the owner of the file). Full tier: `--run-scripts` once in FINISH.
## 10. Finishing
1. `python3 method/gates/post.py --from CONVENER --to ALL --type CLOSE --text "CLOSE: make your files final within 10 minutes, run your board check, log DONE final, then end with your final report"`
and send the ASK lines.
2. Keep looping `wait` until `FINAL-DONE` (every peer logged DONE `--ref final`) or 12 minutes have passed;
`mcp__oivo__sessions({action: "wait_for", sessions: ["<peer ids>"], waitMode: "all", timeout: 110})` tells
you when they are idle.
3. export, manifest, gate. Findings in a peer's file with time left before the close-by time: one NUDGE per
owner with its lines, wait for its DONE, at most one more round. Findings in your own files: fix them.
4. Stop every peer in one message, one `mcp__oivo__sessions({action: "stop", session: "<id>"})` per peer,
then `python3 method/gates/board_check.py roster --status <ROLE>=done ...`; check with `status` that none
is running and that no stand-in subagent of yours is still working.
5. Final export (then the degraded REPORT.md of section 8, if it is yours) and manifest, then
`python3 method/gates/dd_check.py --tier <tier>` once more; it must PASS, otherwise state why in the final
report.
6. End as the RUN CONTRACT's finishing section requires: the typed JSON final report of your Oivo contract
(tier, gate result, roles live or degraded, open and blocked items, paths of REPORT.md and
manifest.json) and `GUI_STATE` with `state: done`.
agents/mesh-standin.md (28 wierszy)
---
name: mesh-standin
description: Stand-in for one MESH peer role during one phase, used only when that role's Oivo session could not be spawned. Reads the board, does the role's phase work in the role's own files, logs every message with post.py. Never messages sessions.
tools: Read, Write, Edit, Bash, Grep, Glob, WebSearch, WebFetch, ToolSearch
model: inherit
---
You stand in for one peer of a MESH design run (no manager; each role owns its files; every message is
logged on the blackboard `board/`). Your prompt names the role code, the phase, its end time and the run
directory. Work only inside the run directory. If WebSearch or WebFetch is deferred, load it with
`ToolSearch` ("select:WebSearch,WebFetch").
1. Read `board/RUN-CONTRACT.md` (binding boundaries and deliverable formats; its finishing section and its
RESTART line apply to the convener only), the role brief named for your role in `board/ROSTER.md` (under
`method/roles/`), `method/PROTOCOL.md`, `method/HYGIENE.md`, `method/POLICY.md` and
`.claude/skills/dd-hw-handbook/SKILL.md`, then `INPUT/brief.md`, `INPUT/requirements.md` and
`INPUT/TOOLCHAIN.md`. The role's own files are its memory: read them.
2. `python3 method/gates/post.py inbox --role <ROLE> --all`: answer what the role owes with
`python3 method/gates/post.py --from <ROLE> ...`. You cannot send messages: ignore the printed ASK lines;
the recipients read the board.
3. Do the role brief's work for this phase, writing only the role's files (`board_check.py owner <path>`
when unsure); publish numbers as facts; run `python3 method/gates/board_check.py --role <ROLE>` and fix
your ERROR lines.
4. Never create files whose names start with REPORT, SUMMARY, FINDINGS or ANALYSIS. `deliverables/REPORT.md`
is written by a session, never by you: the convener moves it to a live peer (`board/ROSTER.md` shows its
owner); skip the role brief's REPORT.md steps.
5. Stop at the phase end time. Log DONE (`--ref phase:<PHASE>`), then end with a short plain summary:
files written, message ids logged, what is still open for the role.
files/method/HYGIENE.md (24 wierszy)
## Ground rules for every agent in this run 1. **Every number has a home and an origin.** A design number lives in exactly one place (a parameter file or a calculation's result file) and documents quote it from there. Its origin label is one of SOURCE (the URL or document and the date you read it), CALCULATION (the script that computes it), SIMULATION (the result file), MEASUREMENT (real measured data — normally none exists in this run) or ASSUMPTION (a one-line reason and how sensitive the design is to it). A number with no origin is deleted, not guessed. 2. **Nothing is invented.** Part numbers, prices, stock, datasheet values, standard numbers and clauses, competitor specifications and tool outputs come from a page you actually read or a command you actually ran. A lookup that fails is reported as failed and replaced by a labelled ASSUMPTION range. 3. **Estimate before you compute.** Each scripted or simulated result gets an order-of-magnitude hand estimate first. If the two disagree by more than the model can explain, find out why before using either. 4. **Prove a route before trusting it.** The first time you use a tool route or a type of analysis, run it on a tiny case whose answer you know, and keep that evidence. 5. **Status words mean what they say.** PASS, PARTIAL, FAIL, BLOCKED, NOT RUN. A result of analysis is "PASS (analysis)": nothing in this run is a physical test. A FAIL stays FAIL until a new check passes. 6. **The product claims only what the design shows.** Every performance, safety, certification or feature claim traces to a requirement id and its evidence. A claim without evidence is removed or marked as an unproven goal. We are designing a real product for honest crowdfunding, not a pitch. 7. **Nobody will answer you.** You decide. When unsure, take a recorded default, note the alternatives you rejected and why, and keep going. Never stop to wait for a person. 8. **Time is scored.** Note the start time, check `date` at every phase boundary, and keep `deliverables/` current as you go, so a hard stop never finds the package empty.
files/method/POLICY.md (67 wierszy)
# Decision policy - what replaces the human in this run Every peer and the convener apply this policy. Nobody outside the run answers questions, approves anything or breaks ties: this file, the owners, the blind votes and SAFE's veto do. ## 1. Who decides - The owner of an artefact (`board/ROSTER.md`) decides about it, with evidence, and records non-trivial decisions in `board/decisions/D-<ROLE>-<nn>.md` (`method/templates/decision.md`). - The system concept is chosen by the blind vote of all peers on ARCH's ballot; ARCH implements the tally. - A contested decision follows the dispute ladder of `method/PROTOCOL.md` section 7. - SAFE holds the safety veto. The convener decides nothing about the design, votes on nothing, approves nothing and edits no peer's file. ## 2. Priority ladder (the higher rule wins; cite it in every ANS, OBJECTION and DECISION it settles) 1. Safety: no unacceptable hazard; protections that work without firmware; inherently safe design first. 2. Requirement compliance: thresholds before targets; a higher requirement weight before a lower one. 3. Unit cost and a retail price that closes with an honest margin. 4. Schedule and buildability: time to a working prototype, parts in stock at two or more suppliers, routes the toolchain supports. 5. Simplicity: fewer parts, modes, custom components and interfaces. ## 3. Vote weights - Concept ballot: every requirement id is a criterion with the weight the input gives it (the tally reads it; nobody sets or tunes it). Extra criteria: RISK `share: 0.15` (technical and delivery risk, every peer rates it), SAFETY `share: 0.10` (rated by SAFE). A share is that criterion's part of the total weight. - Dispute ballot: SAFETY 5, COMPLIANCE 4, COST 3, SCHEDULE 2, SIMPLICITY 1, every voter rates all five. - Knock-out: an option with a predicted threshold miss (with evidence, from REQ or from a rater of that requirement) is out unless all options are out; then the option with the fewest knock-outs wins and its misses are reported as `not_met`. - Tie (equal weighted rating to two decimals): higher SAFETY, then the higher lowest criterion, then the owner's original choice. A winner whose lead flips under the +/-20 % weight sensitivity is still the winner; the decision record names the flipping criteria as risks. ## 4. Defaults for taste questions (when neither the brief nor a requirement decides) - Markets EU and US; SI units, metric ISO fasteners; prices in EUR unless the brief names a currency; owner-facing documents in English. - Prefer architectures that make conformity easy: low voltage inside through a certified external supply when the power level allows; pre-certified radio modules; flame-rated (UL 94 V-0 or V-1) plastics near heat and live parts; hardware protections independent of firmware. - Parts: active lifecycle, in stock at two or more suppliers, the offline catalogue of `INPUT/TOOLCHAIN.md` first; one critical part with a single source needs a named fallback. - Form: compact and calm; matte dark housing with at most one accent colour; one obvious main control plus status indication; no app needed for the basic function unless a requirement asks for it. - Manufacturing: prototype by 3D printing, laser-cut sheet or CNC; 1000 units by moulding or formed sheet metal; walls and drafts chosen for the production process from the start. - Service: standard screws, replaceable wear parts and consumables, no glue where repair matters. - Between two options equal under section 2: the one with fewer custom parts. ## 5. Time plan and stop rules - The clock is `board/CLOCK.json` (start = the convener's first `date`). Concept tier: SETUP to T+10, FRAME to T+28, CONVERGE to T+43, WRITE to T+58, CLOSE to T+63; the convener stops the run by T+80 at the latest whatever the state. Full tier: SETUP to T+10, FRAME to T+30, CONVERGE to T+50, WRITE to T+75, BUILD to T+5:30, CROSSCHECK to T+7:00, FINISH to T+8:30, CLOSE to T+9:00; stop by T+9:30 at the latest. A short hard stop compresses the plan (the clock does it). - Depth: at the concept tier first-order numbers (hand estimate, then a short script); a simulation only if it can flip a decision and fits the phase. At the full tier the handbook depth. - Stop refining a number when it can no longer change a decision or a compliance status. - Behind schedule: cut depth and polish, never honesty, never a compliance row, never the evidence of a top-weighted requirement, never a safety item. - A tool route that fails three times with new information each time: labelled fallback, a BLOCKED or a `blocked[]` entry if it mattered, move on. - Missing information: an ASSUMPTION fact with its reason and sensitivity; never stop to ask anyone. ## 6. Honest claims A claim (performance, safety, certification, feature) exists only with its requirement id and evidence in the deliverables. A predicted miss is reported as a miss. Under-promise.
files/method/PROTOCOL.md (113 wierszy)
# MESH protocol - binding for every peer and for the convener
There is no manager. Each peer owns artefacts and decides about them; peers negotiate directly; every
message, number, idea, decision and vote lives on the blackboard `board/`, the single source of truth. The
convener keeps the clock, runs the checkers and closes the run; it never decides, votes, edits your files
or carries messages between peers. All commands run from the run directory.
## 1. Single writer
- `board/ROSTER.md` lists every role, its session id, status and owned paths; the longest matching pattern
wins. Every peer also owns its board files, `deliverables/02-concept/sections/<ROLE>.md`, and calculation
scripts named `<role>_*.py` in `deliverables/02-concept/calc/` (concept) and `deliverables/05-calculations/`
(full), each writing `<stem>.results.json` and run with `ddpy <script>` from its folder.
- Write only what you own. Unsure? `python3 method/gates/board_check.py owner <path>`. Anything else you want
changed: send a CR to its owner. A file nobody owns is a protocol error: rename it to an owned path.
- Subagents you launch (Agent tool) write only files you own, never message sessions, and cannot create
files named REPORT*, SUMMARY*, FINDINGS* or ANALYSIS*.
## 2. Messages: log first, then send
1. `python3 method/gates/post.py --from <YOU> --to <ROLE>[,<ROLE>]|ALL --type <TYPE> [--re <M-id>] [--ref <path>] --text "..."`
2. It prints one `ASK {...}` line per recipient: pass exactly those fields to `mcp__oivo__sessions` (a
deferred tool: load it once with `ToolSearch` "select:mcp__oivo__sessions"), e.g.
`mcp__oivo__sessions({action: "ask", session: "<id>", message: "<message>", redirect: true})`
(`redirect: true`: a busy peer reads it at its next tool boundary, an idle one is woken; FYI lines carry
`fyi: true` instead: shown at the recipient's next turn, never waking it). Several ASK lines: send them in
one message, in parallel. "no ask needed" or "no live session": the board entry is enough; that role
reads it in its inbox.
| type | use | needs |
|---|---|---|
| Q | question to the owner of a number or file | |
| CR | change request to an owner (what, why, by how much, evidence) | |
| ANS | answer, or reasoned rejection of a CR | `--re` |
| DECISION | owner's decision; accepting a CR is a DECISION `--re` the CR | first `--ref` = the artefact you own |
| OBJECTION | you contest an owner's decision, citing evidence and POLICY.md | `--re` the decision |
| CALL | opens a vote | `--ref board/votes/<YOU>-<topic>/BALLOT.yaml` |
| VOTE | your vote file is written (board only) | `--re` the CALL, `--ref` your vote file |
| VETO / LIFT | SAFE only: safety veto and its lifting | LIFT `--re` the VETO |
| INFO | information the recipients need for their current work, without a request (wakes them) | |
| FYI | a courtesy copy nobody has to act on soon (does not wake anybody) | |
| DONE | your phase work is done (board only) | `--ref phase:<PHASE>` or `--ref final` |
| BLOCKED | something only a human can do (board only) | text `what :: why` |
| PHASE NUDGE TALLY CLOSE | convener only | |
- At most 600 characters; reference files and fact ids instead of pasting; one issue per message (several
CRs to the same owner go in one message); no acknowledgements, thanks or status chatter.
- An incoming message starts with `[M-nnnn TYPE FROM->TO]`. Answer with `--re M-nnnn`.
- The convener decides nothing: post.py refuses a Q, CR, OBJECTION or VETO to CONVENER. A procedural
problem (a script, the roster, the clock, a silent peer) is an INFO `--to CONVENER`; never `ask` the
convener, it reads the board. Never message any session that is not in `board/ROSTER.md`. Your parent
session (the convener) does not answer design questions.
## 3. Answer duty: 15 minutes
A Q, CR or OBJECTION addressed to you is answered within your next work step and at most 15 minutes:
accept (change your artefact and facts, then DECISION `--re`) or reject (ANS `--re` with evidence: fact ids,
calculation results, requirement ids, the POLICY.md rule). A message to ALL is closed by the first useful
answer; a message to named roles needs an answer from each of them. Threads open longer get a NUDGE.
## 4. Never wait idle
Ask, then continue on a labelled ASSUMPTION (a fact with label ASSUMPTION and a note `pending M-nnnn`);
reconcile when the answer arrives. When nothing is actionable, end your turn with one status line: an
incoming message wakes you. Before CLOSE never write your final report (if your Oivo contract wants a
report at the end of every turn, mark it as waiting for the next phase, not as finished).
## 5. Numbers: `board/facts/<ROLE>.yaml` (format: `method/templates/facts.yaml`)
- Every design number you own is a fact `<role>.<name>` with value, unit, label and source (its home).
Numbers of others that you rely on go under `uses:` with the value you used: when the owner changes it,
the checker flags you STALE. Interface agreements are `le:`/`ge:` constraints; budget contributions carry
`budget:` and are summed against `arch.budget.<name>`.
- Key figures have fixed ids: `arch.dimensions` [L, W, H] mm, `arch.mass` kg, `arch.power` W (ARCH);
`cost.unit_cost_q1`, `cost.unit_cost_q100`, `cost.unit_cost_q1000`, `cost.retail_price`, unit = the
currency code (COST). Documents quote facts, never retype them from memory.
- After every change: `python3 method/gates/board_check.py --role <YOU>` and fix your ERROR lines.
## 6. Votes are blind
- A vote has a ballot written by its caller before the CALL (`method/templates/ballot.yaml`); the ballot is
frozen by the CALL (the tally compares hashes). Voters write `board/votes/<topic>/<YOU>.yaml`
(`method/templates/vote.yaml`) BEFORE opening any other file in that folder except the ballot, then log
VOTE. Nobody discusses ratings before the TALLY. Rate every criterion assigned to you (requirement
criteria: `python3 method/gates/board_check.py reqs`; extra criteria: the ballot); others only with evidence.
- A `ko` entry (a predicted threshold miss: the requirement id first, then the predicted value against the
threshold and the fact behind it) knocks an option out when it comes from REQ (the threshold screen) or
from a rater of that requirement in `board/reqmap.yaml`; anybody else's is listed as not counted. SAFE's
`veto` excludes an option. Every section of a vote file is a mapping, as in the template. The convener
runs `board_check.py tally` when all voters are in or the ballot closes, and posts the TALLY; it never
votes. The tally is final; the owner implements it (DECISION).
## 7. Disputes and the safety veto
1. The owner decides about its artefact. 2. Disagree: OBJECTION `--re` the decision with evidence and the
POLICY.md priority that supports you; the owner answers within 15 minutes. 3. Still split: the objector
opens a dispute ballot (`kind: dispute`, the five policy criteria, voters = the affected owners + SAFE) and
CALLs it; the tally binds both sides. 4. SAFE may VETO any decision or option with the hazard, its severity
and the condition for lifting; the owner redesigns or mitigates and SAFE posts LIFT when satisfied. A veto
cannot be outvoted.
## 8. Your work step (repeat until CLOSE)
1. `date`; `python3 method/gates/post.py inbox --role <YOU>` (clock, what you owe, everything new for you
since your last inbox: answers, decisions, INFO, tallies, nudges; your fact errors and stale numbers).
2. Answer what you owe (section 3).
3. Do the work of the current phase (your role brief).
4. Update your facts, run your board check, keep your deliverable files current.
5. Phase work done: DONE `--ref phase:<PHASE>`, then pick up the next phase's work or end your turn.
## 9. Clock and phases
`board/CLOCK.json`; `python3 method/gates/post.py clock` prints the phase, time left and the close-by time.
Phases are time boxes, not handoffs: everybody works at once. At a boundary publish what you have (labelled
honestly), log DONE and move on; unfinished items continue in the next phase. A PHASE message marks each
boundary; when every peer is DONE early, the convener starts the next phase early.
## 10. CLOSE
On CLOSE: within 10 minutes make your owned files consistent with the final facts and decisions, run
`python3 method/gates/board_check.py --role <YOU>`, log DONE `--ref final`, then end with your final report
(Oivo child report). The convener stops your session afterwards.
files/method/gates/board_check.py (855 wierszy)
#!/usr/bin/env python3
"""board_check.py - consistency checker of the MESH blackboard and the convener's tools.
Anyone: [check] [--role R] facts, STALE numbers, le/ge constraints, budgets, ownership, threads (exit 1 on ERROR)
owner PATH... reqs (ids, weights, raters, citing facts) clock
Convener: init --tier concept|full --start ISO --hard-stop-min N [--roles REQ,ARCH,...] [--merge COST:REQ]
roster [--add CODE,...] [--merge SRC:HOST] [--set ROLE=SESSION_ID ...] [--status ROLE=live|degraded|dead|done]
[--move PATH_PATTERN=ROLE] (a merged SRC must not be a roster role: its files go to HOST)
check --write-status [--exit PHASE] wait [--max 240] [--hold: no early return] clock --advance
tally --topic T [--force] export manifest --tier T gate --tier T [--run-scripts]
All as: python3 method/gates/board_check.py <command> ...
"""
import argparse
import csv
import datetime as dt
import glob
import json
import os
import re
import shutil
import statistics
import subprocess
import sys
import time
sys.dont_write_bytecode = True
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import meshlib as M # noqa: E402
LIVE = ("live", "degraded")
WATCH = {"VETO", "OBJECTION", "BLOCKED"}
CORE = ("CONVENER", "REQ", "ARCH", "SAFE", "COST")
KIND = {"01-requirements": "requirements", "02-concept": "concept", "03-cad": "cad", "04-drawings": "drawing",
"05-calculations": "calculation", "06-electronics": "electronics", "07-firmware": "firmware",
"08-safety-certification": "safety", "09-bom-cost": "bom-cost", "10-prototype-test-plan": "test-plan",
"11-renders": "render", "REPORT.md": "report"}
def fail(msg, code=2):
print(f"REFUSED: {msg}")
sys.exit(code)
def files_under(root):
return sorted(M.rel(os.path.join(d, n)) for d, _, ns in os.walk(root) for n in ns if not n.startswith("."))
def ref_topic(m):
for r in m.get("refs", []):
p = r["path"] if isinstance(r, dict) else str(r)
if p.endswith("BALLOT.yaml"):
return os.path.basename(os.path.dirname(p)), (r.get("sha") if isinstance(r, dict) else None)
return None, None
def ballot(topic):
return M.load_ballot(topic)[0]
def closes_at(b):
c = b.get("closes")
if isinstance(c, int): # YAML 1.1 reads an unquoted 11:38 as sexagesimal minutes
c = f"{c // 60:02d}:{c % 60:02d}"
m = re.match(r"^(\d{1,2}):(\d{2})$", str(c or ""))
if not m:
return None
t = M.now().replace(hour=int(m.group(1)), minute=int(m.group(2)), second=0, microsecond=0)
ck = M.clock()
start = M.parse_t(ck["start"]) if ck else None
return t + dt.timedelta(days=1) if start and t < start - dt.timedelta(hours=1) else t
def last_activity(log):
return {m.get("from"): m.get("ts") for m in log}
def done_refs(m):
return [str(r.get("path") if isinstance(r, dict) else r).strip().lower() for r in m.get("refs", [])]
def done_set(log, phase):
return {m["from"] for m in log if m["type"] == "DONE" and f"phase:{phase}".lower() in done_refs(m)}
def final_set(log):
"""Roles that logged DONE --ref final (also accepted: phase:final, phase:CLOSE)."""
return {m["from"] for m in log if m["type"] == "DONE" and {"final", "phase:final", "phase:close"} & set(done_refs(m))}
def exits(phase, ros, log, facts):
"""[(role, missing item)] of a phase's exit list (concept work; full-tier folders are dd_check's job). Items of a
merged role (COST merged into REQ) go to the role that holds it."""
peers, host = [r for r in ros if r != "CONVENER"], M.hosts(ros)
need = []
if phase == "FRAME":
need = [("REQ", "board/reqmap.yaml"), ("REQ", "deliverables/01-requirements/compliance.csv"),
("REQ", "deliverables/01-requirements/requirements.md"), ("ARCH", "board/ideas/ARCH.md"),
("ARCH", "deliverables/02-concept/physics-map.md"), ("ARCH", "deliverables/REPORT.md"),
("COST", "deliverables/02-concept/benchmark.md"), ("SAFE", "board/ideas/SAFE.md")]
need += [(r, f"board/facts/{r}.yaml") for r in peers]
need += [(r, f"2 fragments '## F-{r}-' in board/ideas/{r}.md") for r in peers if r not in CORE]
elif phase == "CONVERGE":
need = [("ARCH", "board/concepts.md"), ("ARCH", "board/votes/ARCH-concept/BALLOT.yaml"),
("ARCH", "CALL of ARCH-concept"), ("ARCH", "DECISION on the concept TALLY")]
elif phase == "WRITE":
need = [(r, f"deliverables/02-concept/sections/{r}.md") for r in peers if r not in ("REQ", "ARCH")]
need += [("ARCH", "deliverables/02-concept/concept.md"), ("ARCH", "deliverables/02-concept/interfaces.md")]
need += [("ARCH", f"fact arch.{k}") for k in ("dimensions", "mass", "power")]
need += [("COST", f"fact cost.{k}") for k in ("unit_cost_q1", "unit_cost_q100", "unit_cost_q1000", "retail_price")]
calls = {m["id"] for m in log if m["type"] == "CALL" and ref_topic(m)[0] == "ARCH-concept"}
anchors = calls | {m["id"] for m in log if m["type"] == "TALLY" and m.get("re") in calls}
out = []
for role, item in need:
p = os.path.join(M.RUN, item.split(" in ")[-1])
role = host.get(role, role)
if role not in ros:
continue
if item.startswith("fact "):
ok = item[5:] in facts
elif item.startswith("2 fragments"):
ok = os.path.isfile(p) and open(p, encoding="utf-8").read().count(f"## F-{role}-") >= 2
elif item.startswith("CALL"):
ok = bool(calls)
elif item.startswith("DECISION"):
ok = any(m["type"] == "DECISION" and m["from"] == "ARCH" and m.get("re") in anchors for m in log)
else:
ok = os.path.isfile(p) and os.path.getsize(p) > 0
if not ok:
out.append((role, item))
return out
# ---------- check ----------
def collect():
ros, log, t = M.roster(), M.read_log(), M.now()
out = [] if ros else [("ERROR", "CONVENER", "board/ROSTER.md missing (board_check.py init)")]
out += [("WARN", "CONVENER", f"{r} has no session yet (roster --set)") for r, i in ros.items() if i["status"] == "pending"]
out += [("ERROR", "CONVENER", f"board/log.jsonl has a broken line ({m['id']}): append only via post.py") for m in log if m["type"] == "BAD"]
for m, age, pend in M.open_threads(log, t):
if m["type"] == "CALL":
continue
who = ["ALL"] if "ALL" in pend else pend
stalled = age > M.STALL_MIN
out += [("WARN" if stalled else "OPEN", r, f"{'STALLED' if stalled else 'open'} {m['id']} {m['type']} "
f"{m['from']}->{','.join(m['to'])} {age:.0f} min: {m['text'][:140]}") for r in who]
facts, probs, budgets = M.facts_check()
out += probs
rprobs = []
M.reqmap(rprobs)
out += [("ERROR", M.hosts(ros).get("REQ", "REQ"), p) for p in rprobs]
for p in files_under(M.DELIV) + files_under(M.BOARD):
owner, cands = M.owner_of(p, ros)
if p != "board/log.jsonl" and not cands:
out.append(("WARN", "CONVENER", f"unowned file {p}: its writer must move it to an owned path"))
elif p != "board/log.jsonl" and not owner:
out.append(("ERROR", "CONVENER", f"ambiguous owner of {p}: {cands} (fix ROSTER.md)"))
for m in log:
topic, s = ref_topic(m) if m["type"] == "CALL" else (None, None)
if m["type"] != "CALL":
continue
b, why = M.load_ballot(topic) if topic else (None, "missing")
if not b:
out.append(("ERROR", m["from"], f"CALL {m['id']}: ballot of topic {topic} {why}"))
elif s and M.sha(os.path.join(M.BOARD, "votes", topic, "BALLOT.yaml")) != s:
out.append(("ERROR", m["from"], f"ballot {topic} changed after its CALL {m['id']}: a ballot is frozen by its CALL"))
return ros, log, facts, budgets, out
def cmd_check(a):
ros, log, facts, budgets, out = collect()
role = a.role.upper() if a.role else None
sel = [o for o in out if not role or o[1] in (role, "ALL")]
print(M.clock_line())
print(f"board: {len(log)} messages, {len(facts)} facts, roles {', '.join(ros)}")
for name, total, unit, lim, n in budgets:
print(f"budget {name}: {n} contributions = {total} {unit}" + (f" (limit {lim} {unit})" if lim is not None else f" (no arch.budget.{name})"))
for sev in ("ERROR", "WARN", "OPEN"):
print("\n".join(f"{sev:5} {o[1]:8} {o[2]}" for o in sel if o[0] == sev) or "", end="\n" if any(o[0] == sev for o in sel) else "")
n_err = sum(o[0] == "ERROR" for o in sel)
print(f"board_check: {'role ' + role if role else 'all roles'}: errors={n_err} warnings={sum(o[0] == 'WARN' for o in sel)} "
f"open={sum(o[0] == 'OPEN' for o in sel)}")
if a.exit:
miss = exits(a.exit.upper(), ros, log, facts)
print(f"EXIT {a.exit.upper()}: " + (f"{len(miss)} item(s) missing - NUDGE each owner with its items" if miss else "complete"))
print("\n".join(f" {r:8} {item}" for r, item in miss) or "", end="\n" if miss else "")
if a.write_status:
ck, act = M.clock(), last_activity(log)
done, final = done_set(log, M.phase_now(ck)[0] if ck else "?"), final_set(log)
L = [f"# STATUS (board_check.py, {M.hm(M.now())})", "", M.clock_line(ck), "",
"| role | session | status | last board activity | DONE this phase | DONE final |", "|---|---|---|---|---|---|"]
L += [f"| {r} | {i['session']} | {i['status']} | {M.hm(M.parse_t(act[r])) if act.get(r) else '-'} | {'yes' if r in done else '-'} "
f"| {'yes' if r in final else '-'} |" for r, i in ros.items()]
L += ["", f"Messages: {len(log)}; facts: {len(facts)}."] + [f"- budget {n}: {tt} {u} from {k} contributions; limit {lim}" for n, tt, u, lim, k in budgets]
L += ["", "## Findings (ERROR: the named owner fixes it; OPEN: a thread awaits its owner)", ""]
L += [f"- {s} {r}: {x}" for s, r, x in out] or ["- none"]
open(os.path.join(M.BOARD, "STATUS.md"), "w", encoding="utf-8").write("\n".join(L) + "\n")
print("wrote board/STATUS.md")
return 1 if n_err else 0
# ---------- init / roster / clock / owner / reqs ----------
def table_rows(lines):
for i, line in enumerate(lines):
cells = [c.strip() for c in line.strip().strip("|").split("|")]
if len(cells) >= 5 and M.ROLE_RE.match(cells[0]):
yield i, cells
def template_rows():
tpl = open(os.path.join(M.RUN, "method", "templates", "ROSTER.md"), encoding="utf-8").read().splitlines()
return tpl, {c[0]: c for _, c in table_rows(tpl)}
def new_row(code, tpl_rows):
if not M.ROLE_RE.match(code) or code in ("CONVENER", "ALL", "DOMAIN"):
fail(f"bad role code {code!r} (2-8 capitals/digits, not CONVENER, ALL or DOMAIN)")
return list(tpl_rows[code]) if code in tpl_rows else [code, "PENDING", "pending", "method/roles/DOMAIN.md", "-"]
def apply_merges(rows, merges, tpl_rows, is_init=False):
"""rows: {role: cells} of the roster (CONVENER excluded). A merged role keeps its brief and paths, held by HOST;
`init` refuses a merged role that is also in --roles, `roster --merge` removes it while it is still pending."""
for src, dst in merges.items():
if src == dst or src not in tpl_rows or src in ("CONVENER", "REQ", "ARCH", "SAFE"):
fail(f"--merge {src}:{dst}: only MECH, ELEC or COST can be merged, into another role")
if src in rows and (rows[src][2] != "pending" or rows[src][1] not in ("PENDING", "-", "")):
fail(f"--merge {src}:{dst}: {src} already has a session; merge only a role nobody runs yet")
if src in rows and is_init:
fail(f"--merge {src}:{dst}: leave {src} out of --roles (its files and facts go to {dst})")
rows.pop(src, None)
if dst not in rows:
fail(f"--merge {src}:{dst}: {dst} is not a roster role")
if f"roles/{src}.md" in " ".join(r[3] for r in rows.values()):
fail(f"--merge {src}:{dst}: {src} is already merged")
src_row = tpl_rows[src]
rows[dst][3] += f" + {src_row[3]}"
rows[dst][4] = ", ".join(x for x in (rows[dst][4], src_row[4]) if x.strip() not in ("", "-")) or "-"
def parse_merges(items):
out = {}
for x in items or []:
src, sep, dst = x.upper().partition(":")
if not sep or not src.strip() or not dst.strip():
fail(f"--merge {x!r}: write SRC:HOST, e.g. COST:REQ")
out[src.strip()] = dst.strip()
return out
def write_roster(head, conv, rows):
lines = "\n".join(head).rstrip() + "\n" + "\n".join("| " + " | ".join(c) + " |" for c in [conv] + list(rows.values())) + "\n"
open(os.path.join(M.BOARD, "ROSTER.md"), "w", encoding="utf-8").write(lines)
def cmd_init(a):
if os.path.isfile(os.path.join(M.BOARD, "CLOCK.json")) and not a.force:
fail("board already initialised (use --force to rewrite CLOCK.json and ROSTER.md; the log is kept; "
"to add or merge roles use `roster --add/--merge`)")
start = M.parse_t(a.start) if a.start else M.now()
if not start:
fail(f"--start {a.start!r} is not ISO 8601 (use the output of `date -Iseconds`)")
roles = [r.strip().upper() for r in a.roles.split(",") if r.strip()]
merges = parse_merges(a.merge)
tpl, tpl_rows = template_rows()
rows = {}
for r in roles:
rows[r] = new_row(r, tpl_rows)
apply_merges(rows, merges, tpl_rows, is_init=True)
cb = M.CLOSE_BY[a.tier]
f = min(1.0, 0.85 * a.hard_stop_min / cb) if a.hard_stop_min else 1.0
at = lambda m: M.iso(start + dt.timedelta(minutes=m * f)) # noqa: E731
ck = {"tier": a.tier, "start": M.iso(start), "scale": round(f, 3), "hard_stop_min": a.hard_stop_min,
"phases": [{"name": n, "end": at(m)} for n, m in M.PLANS[a.tier]], "close_by": at(cb), "advanced": []}
for d in ("board/facts", "board/ideas", "board/votes", "board/decisions", "board/out", "deliverables/01-requirements",
"deliverables/02-concept/sections", "deliverables/02-concept/calc", "deliverables/02-concept/mesh"):
os.makedirs(os.path.join(M.RUN, d), exist_ok=True)
open(M.LOG, "a").close()
json.dump(ck, open(os.path.join(M.BOARD, "CLOCK.json"), "w"), indent=1)
head = [ln for ln in tpl if not any(ln.strip().startswith(f"| {r} |") for r in tpl_rows)]
write_roster(head, tpl_rows["CONVENER"], rows)
print(f"board initialised: tier {a.tier}, roles {', '.join(roles)}" + (f", merged {merges}" if merges else ""))
print_plan(ck)
def print_plan(ck):
print(f"PLAN (local time; start {M.hm(M.parse_t(ck['start']))}; scale {ck.get('scale', 1)}): "
+ " | ".join(f"{p['name']} until {M.hm(M.parse_t(p['end']))}" for p in ck["phases"])
+ f" | close by {M.hm(M.parse_t(ck['close_by']))}")
print(M.clock_line(ck))
def cmd_clock(a):
ck = M.clock() or fail("no board/CLOCK.json (init first)")
if a.advance:
t = M.now()
name, end, _, _ = M.phase_now(ck, t)
if not end:
fail("no running phase to advance")
delta = end - t
for p in ck["phases"]:
if M.parse_t(p["end"]) >= end:
p["end"] = M.iso(M.parse_t(p["end"]) - delta)
ck["advanced"].append({"phase": name, "at": M.iso(t), "saved_min": round(delta.total_seconds() / 60, 1)})
json.dump(ck, open(os.path.join(M.BOARD, "CLOCK.json"), "w"), indent=1)
print(f"advanced: {name} ended now, plan moved {delta.total_seconds() / 60:.0f} min earlier; announce the new phase (PHASE)")
print_plan(ck)
def cmd_roster(a):
path = os.path.join(M.BOARD, "ROSTER.md")
if not os.path.isfile(path):
fail("board/ROSTER.md missing (init first)")
lines = open(path, encoding="utf-8").read().splitlines()
found = list(table_rows(lines))
head, rows = [ln for i, ln in enumerate(lines) if i not in {j for j, _ in found}], {c[0]: c for _, c in found}
_, tpl_rows = template_rows()
conv = rows.pop("CONVENER", None) or list(tpl_rows["CONVENER"])
for code in [c.strip().upper() for x in a.add or [] for c in x.split(",") if c.strip()]:
held = [r for r, c in rows.items() if f"roles/{code}.md" in c[3] and r != code]
if code in rows or held:
fail(f"--add {code}: already in ROSTER.md" + (f" (merged into {held[0]})" if held else ""))
rows[code] = new_row(code, tpl_rows)
apply_merges(rows, parse_merges(a.merge), tpl_rows)
pairs = lambda items: [(k.strip().upper(), v.strip()) for k, _, v in (s.partition("=") for s in items or [])] # noqa: E731
for r, sid in pairs(a.set):
if r not in rows or not sid:
fail(f"--set {r}={sid}: {r} is not a roster role or the session id is empty")
rows[r][1], rows[r][2] = sid, "live"
for r, st in pairs(a.status):
if r not in rows or st.lower() not in ("pending", "live", "degraded", "dead", "done"):
fail(f"--status {r}={st}: unknown role or status (pending, live, degraded, dead, done)")
rows[r][2] = st.lower()
for pat, _, r in (s.rpartition("=") for s in a.move or []):
r = r.strip().upper()
if not pat.strip() or (r not in rows and r != "CONVENER"):
fail(f"--move {pat}={r}: write PATH_PATTERN=ROLE with a roster role or CONVENER")
for c in [conv] + list(rows.values()):
c[4] = ", ".join(p for p in (x.strip() for x in c[4].split(",")) if p and p != "-" and p.strip("`") != pat.strip()) or "-"
tgt = conv if r == "CONVENER" else rows[r]
tgt[4] = ", ".join(p for p in (tgt[4], pat.strip()) if p != "-")
write_roster(head, conv, rows)
for r, i in M.roster().items():
print(f"{r:8} {i['status']:9} {i['session']:38} {i['brief']}")
def cmd_owner(a):
ros = M.roster()
for p in a.paths:
p = os.path.relpath(p, M.RUN) if os.path.isabs(p) else p
owner, cands = M.owner_of(p, ros)
print(f"{p}: {owner or ('AMBIGUOUS ' + ','.join(cands) if cands else 'nobody')}")
def cmd_reqs(_a):
ids, weights = M.requirements()
probs, cite, ros = [], {}, M.roster()
rmap, host = M.reqmap(probs), M.hosts(ros)
for f in M.facts_check()[0].values():
for rid in M.vals(f.get("req") or []):
cite.setdefault(str(rid), []).append(f"{f['id']}={f.get('value')} {f.get('unit')} ({f.get('label')})")
print(f"{len(ids)} requirement ids; weights from INPUT/requirements.md (Priority or Weight field); raters from board/reqmap.yaml")
for p in probs:
print(f"ERROR {p}")
for rid in ids:
print(f"{rid:14} w={weights.get(rid, '?')!s:4} raters={','.join(rmap.get(rid, ['UNMAPPED']))} facts: {'; '.join(cite.get(rid, [])) or '-'}")
if [r for r in ids if r not in weights]:
print(f"WARN no Priority/Weight field line for {', '.join(r for r in ids if r not in weights)}: the tally uses weight 1")
if set(rmap) - set(ids):
print(f"WARN reqmap ids not in INPUT/requirements.md: {', '.join(sorted(set(rmap) - set(ids)))}")
bad = sorted({r for v in rmap.values() for r in v if r != "ALL" and r not in host})
if bad:
print(f"WARN reqmap raters that are not roles of board/ROSTER.md: {', '.join(bad)}")
# ---------- wait ----------
def ready_ballots(log, ros=None):
"""[(topic, cast, voters, key)] of open CALLs whose voters all wrote a vote file or whose ballot has closed. key
changes when a vote file is rewritten or the ballot closes, so a tally refused for an unusable vote file is
signalled again."""
out = []
for m, _, _ in M.open_threads(log):
topic = ref_topic(m)[0] if m["type"] == "CALL" else None
b = ballot(topic) if topic else None
if b:
voters = M.ballot_voters(b, ros)
files = [os.path.join(M.BOARD, "votes", topic, f"{v}.yaml") for v in voters]
cast = [p for p in files if os.path.isfile(p)]
ca = closes_at(b)
closed = bool(ca and M.now() >= ca)
if voters and len(cast) == len(voters) or closed:
key = f"{topic}@{'closed' if closed else 'all'}@{max([int(os.path.getmtime(p)) for p in cast] or [0])}"
out.append((topic, len(cast), len(voters), key))
return out
def cmd_wait(a):
path = os.path.join(M.BOARD, ".wait-state.json")
st = json.load(open(path)) if os.path.isfile(path) else {}
for k, v in (("seen", 0), ("phase", None), ("stalls", []), ("ready", []), ("done", []), ("silent", {}), ("final", False)):
st.setdefault(k, v)
t0 = time.time()
while True:
ck, log, ros, t = M.clock(), M.read_log(), M.roster(), M.now()
new = [m for m in log if M.idn(m) > st["seen"]]
key = [m for m in new if m["from"] != "CONVENER" and (m["type"] in WATCH or ("CONVENER" in m["to"] and m["type"] not in ("DONE", "VOTE")))]
phase = M.phase_now(ck, t)[0] if ck else None
threads = M.open_threads(log, t)
stalls = [(m, g, p) for m, g, p in threads if g > M.STALL_MIN and m["type"] != "CALL" and "ALL" not in p and m["id"] not in st["stalls"]]
ready = [r for r in ready_ballots(log, ros) if r[3] not in st["ready"]]
peers = [r for r, i in ros.items() if r != "CONVENER" and i["status"] in LIVE]
pdone = bool(peers) and set(peers) <= done_set(log, phase)
fdone = bool(peers) and set(peers) <= final_set(log) and not st["final"]
act, silent = last_activity(log), []
for r in peers:
last = M.parse_t(act.get(r) or (ck or {}).get("start", ""))
owed = [m for m, g, p in threads if M.pending_for(p, m, r) and m["type"] in ("Q", "CR", "OBJECTION")]
rep = M.parse_t(st["silent"].get(r, "")) if st["silent"].get(r) else None
if last and M.minutes(last, t) > M.STALL_MIN and owed and (not rep or M.minutes(rep, t) > M.STALL_MIN):
silent.append((r, M.minutes(last, t), len(owed)))
ev = [e for e, c in (("messages", key), ("phase", phase != st["phase"] and st["phase"] is not None), ("stalled", stalls),
("votes-ready", ready), ("phase-done", pdone and phase not in st["done"]), ("final-done", fdone),
("silent-peer", silent)) if c]
if (ev and not a.hold) or time.time() - t0 >= a.max:
break
time.sleep(min(10, max(1, a.max - (time.time() - t0))))
print(f"WAIT {time.time() - t0:.0f} s -> {', '.join(ev) or 'no event'}")
print(M.clock_line(ck))
if "phase" in ev and phase == "OVERTIME":
print("PHASE is now OVERTIME (the plan's last box ended): post no PHASE; finish MAIN.md section 10 by the close-by time")
elif "phase" in ev and phase == "CLOSE":
print(f"PHASE is now CLOSE (was {st['phase']}): post no PHASE: start MAIN.md section 10 (the CLOSE message); first "
f"python3 method/gates/board_check.py check --write-status --exit {st['phase']}")
elif "phase" in ev:
print(f"PHASE is now {phase} (was {st['phase']}): post PHASE to ALL (unless you just did), then: "
f"python3 method/gates/board_check.py check --write-status --exit {st['phase']}")
if new:
kinds = {}
for m in new:
kinds[m["type"]] = kinds.get(m["type"], 0) + 1
print(f"BOARD +{len(new)} entries ({', '.join(f'{k} {v}' for k, v in sorted(kinds.items()))})" + (f"; {len(key)} need your attention:" if key else ""))
for m in key[-12:]:
print(f" {m['id']} {M.hm(M.parse_t(m['ts']))} {m['type']} {m['from']}->{','.join(m['to'])}: {m['text'][:200]}")
for m, g, p in stalls:
what = "redesign or mitigate until SAFE lifts it" if m["type"] == "VETO" else "answer"
print(f"STALLED {m['id']} {m['type']} {m['from']}->{','.join(m['to'])} {g:.0f} min, waiting for {','.join(p)} to {what}: "
f"NUDGE them (procedure only, never content)")
for topic, n, tot, _ in ready:
print(f"VOTES READY {topic} {n}/{tot}: python3 method/gates/board_check.py tally --topic {topic}"
+ ("" if n == tot else " (closing time passed: it tallies the votes that are in)"))
if "phase-done" in ev:
print(f"PHASE-DONE every live peer logged DONE for {phase}: python3 method/gates/board_check.py clock --advance, then PHASE")
if "final-done" in ev:
print("FINAL-DONE every live peer logged DONE final: go on with MAIN.md section 10 step 3")
for r, g, n in silent:
print(f"SILENT {r}: no board activity for {g:.0f} min with {n} open request(s): sessions status (fast), then MAIN.md "
f"section 8 (busy: NUDGE only, never stop it)")
st.update(seen=max([M.idn(m) for m in log] or [0]), phase=phase, final=st["final"] or "final-done" in ev)
st["stalls"] += [m["id"] for m, _, _ in stalls]
st["ready"] += [r[3] for r in ready]
st["done"] += [phase] if "phase-done" in ev else []
st["silent"].update({r: M.iso(M.now()) for r, _, _ in silent})
json.dump(st, open(path, "w"))
# ---------- tally ----------
SHAPE = {"ratings": "criterion -> {option: 0..10}", "reasons": "criterion -> one line",
"ko": "option -> 'REQ-ID: predicted value vs threshold (fact id)'", "veto": "option -> 'hazard, severity, lift condition'"}
def as_map(x, what, who, rp, keys, issues):
"""A vote section as a mapping. A list of one-key mappings (or of 'KEY: text' strings with known keys) is read as
one mapping; any other shape is ignored. Both are reported in `issues` with the voter and the file."""
if x in (None, [], {}) or (isinstance(x, str) and x.strip().lower() in ("", "none", "-", "n/a", "no")):
return {}
if isinstance(x, dict):
return x
if isinstance(x, list):
out = {}
for e in x:
if isinstance(e, dict) and len(e) == 1:
out.update(e)
elif isinstance(e, str) and e.partition(":")[0].strip() in keys:
out[e.partition(":")[0].strip()] = e.partition(":")[2].strip()
else:
break
else:
issues.append(f"{who}: `{what}` in {rp} is a list; read as a mapping {SHAPE[what]}")
return out
issues.append(f"{who}: `{what}` in {rp} is not a mapping ({SHAPE[what]}): ignored")
return {}
def read_vote(v, path, crit, opts, issues):
"""({section: mapping}, None) or (None, problem) for one vote file; never raises on its content."""
rp = M.rel(path)
try:
d = M.load_yaml(path)
except Exception as exc: # noqa: BLE001
return None, f"{rp} does not parse: {str(exc).splitlines()[0][:120]}"
if not isinstance(d, dict):
return None, f"{rp} is {'empty' if d is None else 'a ' + type(d).__name__}, not a mapping (method/templates/vote.yaml)"
if d.get("voter") and str(d["voter"]).strip().upper() != v:
issues.append(f"{v}: {rp} says voter {d['voter']!r}; counted as {v} (the file name decides)")
return {k: as_map(d.get(k), k, v, rp, set(crit) if k in ("ratings", "reasons") else set(opts), issues) for k in SHAPE}, None
def cmd_tally(a):
topic, log, ros = a.topic, M.read_log(), M.roster()
b, why = M.load_ballot(topic)
if not b:
fail(f"board/votes/{topic}/BALLOT.yaml {why}")
call = next((m for m in log if m["type"] == "CALL" and ref_topic(m)[0] == topic), None)
if call and any(m["type"] == "TALLY" and m.get("re") == call["id"] for m in log):
fail(f"{topic} is already tallied; a tally is final (a new question needs a new topic)")
issues, host = [], M.hosts(ros)
crit, voters = M.criteria(b, issues), M.ballot_voters(b, ros)
listed = b.get("voters") if isinstance(b.get("voters"), list) else []
issues += [f"ballot voter {x} is not a role of board/ROSTER.md: left out" for x in listed if str(x).upper() not in host and str(x).upper() != "ALL"]
raw = b.get("options") or {}
if isinstance(raw, list):
raw = {k: w for e in raw for k, w in (e.items() if isinstance(e, dict) else [(e, "")])}
opts = {str(k): str(w) for k, w in raw.items()} if isinstance(raw, dict) else {}
if not opts or not voters:
fail(f"ballot {topic} needs options (`id: one line`) and voters (ALL or roles of board/ROSTER.md)")
votes, broken = {}, {}
for v in voters:
p = os.path.join(M.BOARD, "votes", topic, f"{v}.yaml")
if not os.path.isfile(p):
continue
sec, prob = read_vote(v, p, crit, opts, issues)
if prob:
broken[v] = prob
else:
votes[v] = sec
if call and os.path.getmtime(p) < M.parse_t(call["ts"]).timestamp():
issues.append(f"{v}'s vote file predates the CALL {call['id']}")
missing = [v for v in voters if v not in votes and v not in broken]
ca = closes_at(b)
if (missing or broken) and not a.force and not (ca and M.now() >= ca):
parts = ([f"waiting for {', '.join(missing)}"] if missing else []) + \
[f"unusable vote file {x}: NUDGE {v} to rewrite it from method/templates/vote.yaml" for v, x in broken.items()]
fail("; ".join(parts) + f" (closes {M.hm(ca) if ca else '?'}; then the tally counts the usable votes; --force tallies now)", 3)
issues += [f"{v}'s vote is not counted: {x}" for v, x in broken.items()]
req_ids, rmap = M.requirements()[0], M.reqmap()
rat, ko, cko, veto, by = {c: {o: [] for o in opts} for c in crit}, {}, {}, {}, {}
for v, d in votes.items():
for c, row in d["ratings"].items():
if str(c) not in crit or not isinstance(row, dict):
issues.append(f"{v}: unknown criterion {c}" if str(c) not in crit else f"{v}: ratings of {c} must map option -> 0..10")
continue
for o, x in row.items():
if str(o) in opts and M.num(x) and 0 <= x <= 10:
rat[str(c)][str(o)].append(x)
by.setdefault((str(c), str(o)), []).append((v, x))
else:
issues.append(f"{v}: bad rating {c}/{o}={x!r} (options {list(opts)}, 0..10)")
for o, text in ((str(o), str(t)) for o, t in d["ko"].items()):
if o not in opts:
issues.append(f"{v}: ko of unknown option {o}")
continue
# a knock-out counts from REQ (the threshold screen) or from a rater of the requirement it cites (reqmap)
rid = min(((text.find(r), r) for r in req_ids if r in text), default=(0, None))[1]
raters = [host.get(r, r) for r in (rmap.get(rid) or ["REQ"])] if rid else []
if v == host.get("REQ", "REQ") or v in raters or "ALL" in raters:
ko.setdefault(o, []).append(f"{v}: {text}")
else:
cko.setdefault(o, []).append(f"{v}: {text}" + (f" (raters of {rid}: {','.join(raters)})" if rid else " (cites no requirement id)"))
for o, text in ((str(o), str(t)) for o, t in d["veto"].items()):
if v != "SAFE":
issues.append(f"{v} cannot veto (SAFE only)")
elif o not in opts:
issues.append(f"{v}: veto of unknown option {o}")
else:
veto.setdefault(o, []).append(f"{v}: {text}")
issues += [f"{v} did not rate {c}, which is assigned to it" for c, s in crit.items()
if (v in s["roles"] or "ALL" in s["roles"]) and not d["ratings"].get(c)]
rated = [c for c in crit if any(rat[c][o] for o in opts)]
mean = {c: {o: statistics.mean(rat[c][o]) if rat[c][o] else None for o in opts} for c in rated}
def totals(w):
res = {}
for o in opts:
pairs = [(w[c], mean[c][o]) for c in rated if mean[c][o] is not None]
res[o] = sum(x * y for x, y in pairs) / sum(x for x, _ in pairs) if pairs and sum(x for x, _ in pairs) else 0.0
return res
base = {c: crit[c]["weight"] for c in rated}
tot = totals(base)
safety = {o: (mean.get("SAFETY", {}).get(o) or 0) for o in opts}
minr = {o: min([mean[c][o] for c in rated if mean[c][o] is not None] or [0]) for o in opts}
alive = [o for o in opts if o not in ko and o not in veto]
pool = alive or [o for o in opts if o not in veto] or list(opts)
if not alive:
fewest = min(len(ko.get(o, [])) for o in pool)
pool = [o for o in pool if len(ko.get(o, [])) == fewest]
order = lambda tt: (lambda o: (round(tt[o], 2), safety[o], minr[o])) # noqa: E731
rank = sorted(pool, key=order(tot), reverse=True)
win, flips = rank[0], []
for c in rated if len(pool) > 1 else []:
for fct in (0.8, 1.2):
tt = totals(dict(base, **{c: base[c] * fct}))
best = max(pool, key=order(tt))
if best != win:
flips.append(f"{c} x{fct} -> {best}")
if len(pool) < 2:
sens, sens_short = "not applicable: only one option survived the knock-outs and vetoes", "n/a (one surviving option)"
else:
sens = f"winner changes in {len(flips)} case(s): {'; '.join(flips[:12])}" if flips else "winner holds in every case"
sens_short = f"{len(flips)} flips" if flips else "holds"
contested = [f"{c}/{o}: " + ", ".join(f"{v}={x}" for v, x in lst) for (c, o), lst in sorted(by.items())
if len(lst) > 1 and max(x for _, x in lst) - min(x for _, x in lst) >= 4]
cur = M.sha(os.path.join(M.BOARD, "votes", topic, "BALLOT.yaml"))
csha = ref_topic(call)[1] if call else None
lead = tot[rank[0]] - tot[rank[1]] if len(rank) > 1 else None
L = [f"# Tally - {topic}", "", f"Tallied {M.iso(M.now())} by method/gates/board_check.py (run by the convener, who never votes).",
f"Question: {b.get('question', '')}", f"Kind: {b.get('kind', '?')}; criteria: {len(crit)} ({len(rated)} rated); requirement weights "
f"from INPUT/requirements.md; ballot sha {cur}" + (f", CALL {call['id']} sha {csha} ({'unchanged' if csha == cur else 'CHANGED AFTER CALL'})" if call else ", no CALL logged"),
f"Voters: {len(votes)}/{len(voters)}" + (f"; missing: {', '.join(missing)}" if missing else "") + (f"; unusable: {', '.join(broken)}" if broken else ""), "",
"| option | weighted rating (0-10) | knock-outs | vetoes | lowest criterion mean | SAFETY |", "|---|---|---|---|---|---|"]
for o in sorted(opts, key=lambda o: -tot[o]):
mark = " (vetoed)" if o in veto else " (knocked out)" if o in ko else ""
L.append(f"| {o}{mark} {opts[o][:60]} | {tot[o]:.2f} | {len(ko.get(o, []))} | {len(veto.get(o, []))} | {minr[o]:.1f} | {safety[o]:.1f} |")
L += ["", f"**Result: {win}**" + (f" leads by {lead:.2f}" if lead is not None else "")
+ ("" if alive else " (every option has a knock-out: the one with the fewest is taken; its misses are not_met)"),
f"Sensitivity (each criterion weight x0.8 and x1.2, {2 * len(rated)} cases): {sens}",
"Contested ratings (spread >= 4 between voters): " + ("; ".join(contested[:15]) or "none"),
"Unrated criteria: " + (", ".join(c for c in crit if c not in rated) or "none"), ""]
L += [f"- KO {o}: " + " | ".join(x) for o, x in ko.items()] + [f"- VETO {o}: " + " | ".join(x) for o, x in veto.items()]
L += [f"- KO not counted (neither REQ nor a rater of the cited requirement in board/reqmap.yaml) {o}: " + " | ".join(x) for o, x in cko.items()]
L += ["", "| criterion | weight | raters | " + " | ".join(opts) + " |", "|---|---|---|" + "---|" * len(opts)]
L += [f"| {c} | {crit[c]['weight']:.2f} | {','.join(crit[c]['roles'])} | "
+ " | ".join(f"{mean[c][o]:.1f}" if c in mean and mean[c][o] is not None else "-" for o in opts) + " |" for c in crit]
L += ["", "## Votes (revealed after the tally)", ""]
for v, d in votes.items():
L += [f"### {v}"] + [f"- {c}: {d['ratings'].get(c, '')} - {d['reasons'].get(c, '')}" for c in dict.fromkeys(list(d["ratings"]) + list(d["reasons"]))] + [""]
L += (["## Issues", ""] + [f"- {x}" for x in issues]) if issues else []
out = os.path.join(M.BOARD, "votes", topic, "TALLY.md")
open(out, "w", encoding="utf-8").write("\n".join(L) + "\n")
print("\n".join(L[:14 + len(opts)]) + f"\nwrote {M.rel(out)}")
if issues:
print(f"ISSUES ({len(issues)}, listed in TALLY.md; procedural only, the tally stands):\n " + "\n ".join(issues[:12]))
entry = {"from": "CONVENER", "to": voters, "type": "TALLY", "refs": [{"path": M.rel(out), "sha": M.sha(out)}],
"text": f"{topic}: {win} wins" + (f" by {lead:.2f}" if lead is not None else "") + f"; sensitivity {sens_short}; "
f"KO {sorted(ko)}; vetoed {sorted(veto)}. See board/votes/{topic}/TALLY.md"}
entry = M.append_log(dict(entry, re=call["id"]) if call else entry)
print(f"logged {entry['id']} TALLY")
print("\n".join(M.ask_lines(entry, M.roster())))
# ---------- export / manifest ----------
def cell(x, n=160):
return " ".join(str(x).split()).replace("|", "/")[:n]
def row(m, n=260):
return f"| {m['id']} | {M.hm(M.parse_t(m['ts']))} | {m['type']} | {m['from']} -> {','.join(m['to'])} | {cell(m['text'], n)} | {m.get('re', '-')} |"
def cmd_export(_a):
out = os.path.join(M.DELIV, "02-concept", "mesh")
os.makedirs(out, exist_ok=True)
ros, log, facts, budgets, found = collect()
t, replies, count = M.now(), {}, {}
for m in log:
replies.setdefault(m.get("re"), []).append(m)
count[m["type"]] = count.get(m["type"], 0) + 1
hdr = "| id | time | type | from -> to | text | re |\n|---|---|---|---|---|---|"
L = ["# Decision log of the peer mesh", "", f"Generated {M.iso(t)} by `method/gates/board_check.py export` from the board log "
f"({len(log)} messages: " + ", ".join(f"{k} {v}" for k, v in sorted(count.items())) + "). Every message between peers was "
"logged before it was sent; the raw log is `log.jsonl` in this folder.", "", "## Roster", "", "| role | brief | status |", "|---|---|---|"]
L += [f"| {r} | {i['brief']} | {i['status']} |" for r, i in ros.items()]
L += ["", "## Decisions (`re` = the CR, question, objection or tally it settles)", "", hdr] + [row(m, 320) for m in log if m["type"] == "DECISION"]
crs, times, acc, rej, rows = [m for m in log if m["type"] == "CR"], [], 0, 0, []
for m in crs:
rep = [r for r in replies.get(m["id"], []) if r["type"] in ("DECISION", "ANS")]
outcome = "OPEN"
if rep:
r0, dtm = rep[0], M.minutes(M.parse_t(m["ts"]), M.parse_t(rep[0]["ts"]))
acc, rej = acc + (r0["type"] == "DECISION"), rej + (r0["type"] == "ANS")
times.append(dtm)
outcome = f"{'accepted' if r0['type'] == 'DECISION' else 'answered/rejected'} by {r0['from']} in {dtm:.0f} min ({r0['id']}): {cell(r0['text'])}"
rows.append(f"| {m['id']} | {M.hm(M.parse_t(m['ts']))} | {m['from']} -> {','.join(m['to'])} | {cell(m['text'], 220)} | {outcome} |")
L += ["", f"## Change requests ({len(crs)}: {acc} accepted, {rej} answered or rejected, {len(crs) - acc - rej} open"
+ (f"; median answer time {statistics.median(times):.0f} min" if times else "") + ")", "",
"| id | time | from -> to | request | outcome |", "|---|---|---|---|---|"] + rows
L += ["", "## Objections, vetoes, lifts, calls and tallies", "", hdr] + [row(m) for m in log if m["type"] in ("OBJECTION", "VETO", "LIFT", "CALL", "TALLY")]
L += ["", "## Blocked (needs a human)", ""] + ([f"- {m['id']} {m['from']}: {cell(m['text'], 300)}" for m in log if m["type"] == "BLOCKED"] or ["- none"])
L += ["", "## Still open at export", ""] + ([f"- {m['id']} {m['type']} {m['from']}->{','.join(m['to'])} ({g:.0f} min, waiting for "
f"{','.join(p)}): {cell(m['text'], 200)}" for m, g, p in M.open_threads(log, t)] or ["- none"])
V = ["# Votes of the peer mesh", "", "Blind votes: each voter wrote its ratings before seeing the others'; the tally script applies "
"knock-outs, the safety veto, the requirement weights of the input and a +/-20 % weight sensitivity.", ""]
for bp in sorted(glob.glob(os.path.join(M.BOARD, "votes", "*", "BALLOT.yaml"))):
tp = os.path.join(os.path.dirname(bp), "TALLY.md")
V += [open(tp, encoding="utf-8").read().replace("# Tally", "## Tally", 1) if os.path.isfile(tp) else f"## {os.path.basename(os.path.dirname(bp))}: not tallied", ""]
F = ["# Key numbers register (merged board facts)", "", f"Generated {M.iso(t)}. Each number has one owner; `source` is its home "
"(script, results file, URL with date, or the reason of an ASSUMPTION).", "",
"| id | value | unit | label | source | owner | requirement ids |", "|---|---|---|---|---|---|---|"]
F += [f"| {k} | {cell(f.get('value'), 40)} | {cell(f.get('unit'), 20)} | {f.get('label')} | {cell(f.get('source'), 140)} | {f['owner']} | "
f"{cell(', '.join(map(str, M.vals(f.get('req') or []))), 80)} |" for k, f in sorted(facts.items(), key=lambda kv: (kv[1]["owner"], kv[0]))]
F += ["", "## Interface constraints and budgets", ""] + ([f"- {k} {op} {f[op]}" for k, f in sorted(facts.items()) for op in ("le", "ge") if f.get(op)]
or ["- no interface constraints declared"])
F += [f"- budget {n}: {tt} {u} from {k} contributions; limit {lim}" for n, tt, u, lim, k in budgets]
F += ["", "## Consistency findings at export", ""] + ([f"- {s} {r}: {x}" for s, r, x in found if s in ("ERROR", "WARN")] or ["- none"])
G = ["# Concept fragments posted by the peers", "", "Each domain owner posted its own fragments; the system alternatives were "
"assembled from them (see ../concept.md).", ""]
for p in sorted(glob.glob(os.path.join(M.BOARD, "ideas", "*.md"))) + [os.path.join(M.BOARD, "concepts.md")]:
if os.path.isfile(p):
G += [f"<!-- {M.rel(p)} -->", open(p, encoding="utf-8").read().replace("\n# ", "\n## ").strip(), ""]
for name, lines in (("decision-log.md", L), ("votes.md", V), ("facts.md", F), ("fragments.md", G)):
open(os.path.join(out, name), "w", encoding="utf-8").write("\n".join(lines) + "\n")
if os.path.isfile(M.LOG):
shutil.copyfile(M.LOG, os.path.join(out, "log.jsonl"))
print(f"exported decision-log.md, votes.md, facts.md, fragments.md, log.jsonl to {M.rel(out)}/")
def describe(path):
ext, full = os.path.splitext(path)[1].lower(), os.path.join(M.DELIV, path)
if ext in (".md", ".csv", ".py"):
text = open(full, encoding="utf-8", errors="replace").read(800)
if ext == ".py":
m = re.search(r'"""(.+?)(?:\n|""")', text, re.S)
return "script: " + cell(m.group(1) if m else os.path.basename(path), 110)
first = next((ln for ln in text.splitlines() if ln.strip()), "")
return ("table: " if ext == ".csv" else "") + cell(first.lstrip("# "), 120)
if path.endswith(".results.json"):
return "results of " + os.path.basename(path).replace(".results.json", ".py")
return {".json": "JSON data", ".jsonl": "board message log (JSON lines)", ".step": "STEP model", ".stp": "STEP model",
".pdf": "PDF document", ".svg": "SVG drawing", ".png": "PNG image", ".glb": "glTF binary model"}.get(ext, f"{ext.lstrip('.') or 'a'} file")
def cmd_manifest(a):
path = os.path.join(M.DELIV, "manifest.json")
try:
old = {f["path"]: f for f in json.load(open(path, encoding="utf-8")).get("files", [])}
except Exception: # noqa: BLE001
old = {}
files = []
for p in files_under(M.DELIV):
rp = os.path.relpath(os.path.join(M.RUN, p), M.DELIV)
if rp != "manifest.json" and os.path.getsize(os.path.join(M.RUN, p)) > 0:
o = old.get(rp, {})
files.append({"path": rp, "kind": o.get("kind") or KIND.get(rp.split("/")[0], "other"), "desc": o.get("desc") or describe(rp)})
facts = M.facts_check()[0]
def fig(fid, unit):
f = facts.get(fid) or {}
n = len(M.vals(f.get("value")))
pairs = M.compare(f.get("value"), f.get("unit"), [1] * n, unit) if f else None
res = [round(x / y, 4) for x, y in pairs] if pairs else None
return res if res is None or isinstance(f.get("value"), list) else res[0]
cur = (facts.get("cost.unit_cost_q1") or {}).get("unit")
uc = {"currency": cur, **{q: (facts.get(f"cost.unit_cost_{q}") or {}).get("value") if (facts.get(f"cost.unit_cost_{q}") or {}).get("unit") == cur
else None for q in ("q1", "q100", "q1000")}} if cur else None
rp, dims = facts.get("cost.retail_price"), fig("arch.dimensions", "mm")
blocked = [{"what": m["text"].partition("::")[0].strip(), "why": (m["text"].partition("::")[2].strip() or f"see board log {m['id']}") + f" ({m['from']})"}
for m in M.read_log() if m["type"] == "BLOCKED"]
kf = {"dimensions_mm": dims if isinstance(dims, list) and len(dims) == 3 else None, "mass_kg": fig("arch.mass", "kg"),
"power_w": fig("arch.power", "W"), "unit_cost": uc,
"retail_price": {"currency": rp.get("unit"), "value": rp.get("value")} if rp and M.num(rp.get("value")) else None}
json.dump({"tier": a.tier, "files": files, "key_figures": kf, "blocked": blocked}, open(path, "w", encoding="utf-8"), indent=1, ensure_ascii=False)
print(f"wrote deliverables/manifest.json: {len(files)} files, key_figures {json.dumps(kf)}, blocked {len(blocked)}")
if None in kf.values():
print(f"NOTE null key figures {[k for k, v in kf.items() if v is None]}: their owners publish arch.dimensions [L,W,H] mm, arch.mass kg, "
"arch.power W, cost.unit_cost_q1/q100/q1000 and cost.retail_price (unit = currency) as facts")
# ---------- gate ----------
BARE = {"compliance.csv": "01-requirements/compliance.csv", "bom.csv": "09-bom-cost/bom.csv", "cost.json": "09-bom-cost/cost.json",
"cost.md": "09-bom-cost/cost.md"}
TOKEN = re.compile(r"(?:deliverables/)?((?:[01]\d-[\w-]+/?[\w./-]*)|REPORT\.md|manifest\.json|compliance\.csv|bom\.csv|cost\.json|cost\.md)")
def route(msg, ros):
for tok in TOKEN.findall(msg):
cand = "deliverables/" + BARE.get(tok, tok).rstrip("/.")
owner = M.owner_of(cand, ros)[0] or M.owner_of(cand + "/x", ros)[0]
if owner:
return owner
return "ALL"
def integration(facts):
out, names = [], []
asm = os.path.join(M.DELIV, "03-cad", "assembly.step")
if os.path.isfile(asm) and shutil.which("step-info"):
try:
info = json.loads(subprocess.run(["step-info", asm], capture_output=True, text=True, timeout=300).stdout)
size, dims = info.get("size_mm"), (facts.get("arch.dimensions") or {}).get("value")
if size and isinstance(dims, list) and len(dims) == 3 and any(abs(x - y) > 0.03 * max(y, 1) for x, y in zip(sorted(size), sorted(dims))):
out.append(("ERROR", "ARCH", f"03-cad/assembly.step size {size} mm differs from arch.dimensions {dims} mm by > 3 %"))
if not info.get("all_valid", True):
out.append(("ERROR", "ARCH", "03-cad/assembly.step contains invalid shapes (step-info)"))
plist = [p for p in info.get("part_list") or [] if isinstance(p, dict)]
# leaf part names: `product` (CadQuery `name` is '<assembly id>/<part>'); never the assembly roots themselves
roots = {str(p.get("name", "")).split("/")[0] for p in plist if "/" in str(p.get("name", ""))}
names = [str(p.get("product") or str(p.get("name", "")).rsplit("/", 1)[-1]) for p in plist]
names = [n for n in names if n not in roots and n.lower() not in ("unnamed", "")]
except (ValueError, subprocess.TimeoutExpired, OSError) as exc:
out.append(("WARN", "ARCH", f"step-info on assembly.step failed: {exc}"))
pc, bc = os.path.join(M.DELIV, "03-cad", "parts.csv"), os.path.join(M.DELIV, "09-bom-cost", "bom.csv")
if os.path.isfile(bc):
bom = open(bc, encoding="utf-8", errors="replace").read().lower()
parts = [r.get("part", "") for r in csv.DictReader(open(pc, encoding="utf-8", errors="replace"))] if os.path.isfile(pc) else []
cost = M.hosts().get("COST", "COST")
out += [("WARN", cost, f"CAD part '{p}' has no line in bom.csv (description, mpn or notes): add it or tell MECH/ARCH")
for p in sorted(set(x for x in parts + names if x)) if p.lower() not in bom]
return out
def cmd_gate(a):
ros = M.roster()
cmd = [sys.executable, os.path.join(os.path.dirname(os.path.abspath(__file__)), "dd_check.py"), "--tier", a.tier, "--json"]
proc = subprocess.run(cmd + (["--run-scripts"] if a.run_scripts else []), cwd=M.RUN, capture_output=True, text=True, timeout=3600)
try:
res = json.loads(proc.stdout.strip().splitlines()[-1])
except (ValueError, IndexError):
fail(f"dd_check.py gave no JSON: {proc.stdout[-300:]} {proc.stderr[-300:]}")
_, _, facts, _, found = collect()
lines = [("ERROR", route(e, ros), e) for e in res["errors"]] + [("WARN", route(w, ros), w) for w in res["warnings"]]
lines += (integration(facts) if a.tier == "full" else []) + [(s, r, "board: " + x) for s, r, x in found if s == "ERROR"]
by = {}
for s, r, x in lines:
by.setdefault(r, []).append(f"{s} {x}")
print(f"GATE tier={a.tier}: dd_check {'PASS' if res['pass'] else 'FAIL'} ({len(res['errors'])} errors, {len(res['warnings'])} warnings); "
f"{sum(s == 'ERROR' for s, _, _ in lines)} errors in total")
for r in sorted(by):
print(f"[{r}] " + ("owner of the file" if r != "ALL" else "shared folder: every peer that writes there") + "\n " + "\n ".join(by[r]))
print("Route each owner's lines to it with one NUDGE (post.py --from CONVENER --to <ROLE> --type NUDGE ...); never fix a peer's file.")
return 0 if res["pass"] else 1
def main():
ap = argparse.ArgumentParser(description="MESH board checker")
sub = ap.add_subparsers(dest="cmd")
spec = {"check": [("--role", {}), ("--write-status", {"action": "store_true"}), ("--exit", {})],
"init": [("--tier", {"choices": ["concept", "full"], "required": True}), ("--start", {}), ("--hard-stop-min", {"type": float}),
("--roles", {"default": "REQ,ARCH,MECH,ELEC,SAFE,COST"}), ("--merge", {"action": "append"}), ("--force", {"action": "store_true"})],
"roster": [("--add", {"action": "append"}), ("--merge", {"action": "append"}), ("--set", {"action": "append"}),
("--status", {"action": "append"}), ("--move", {"action": "append"})], "owner": [("paths", {"nargs": "+"})],
"reqs": [], "clock": [("--advance", {"action": "store_true"})], "wait": [("--max", {"type": float, "default": 240}), ("--hold", {"action": "store_true"})],
"tally": [("--topic", {"required": True}), ("--force", {"action": "store_true"})], "export": [],
"manifest": [("--tier", {"choices": ["concept", "full"], "required": True})],
"gate": [("--tier", {"choices": ["concept", "full"], "required": True}), ("--run-scripts", {"action": "store_true"})]}
for name, args in spec.items():
p = sub.add_parser(name)
for flag, kw in args:
p.add_argument(flag, **kw)
argv = sys.argv[1:]
a = ap.parse_args(["check"] + argv if not argv or argv[0].startswith("-") else argv)
sys.exit(globals()["cmd_" + a.cmd](a) or 0)
if __name__ == "__main__":
main()
files/method/gates/dd_check.py (431 wierszy)
#!/usr/bin/env python3
"""dd_check.py - check deliverables/ against the RUN CONTRACT layout (contract v1.x).
Usage (from the run directory):
python3 method/gates/dd_check.py --tier concept|full [--run-scripts] [--json]
Checks the formats the RUN CONTRACT fixes (compliance.csv, manifest.json, bom.csv, cost.json, the
folder layout of the tier, evidence paths) and prints ERROR / WARN lines. Exit code 0 = no errors.
It checks form, not engineering quality. Standard library only.
"""
import argparse
import csv
import io
import json
import os
import re
import shutil
import subprocess
import sys
COMPLIANCE_HEADER = ["req_id", "status", "value", "unit", "evidence", "note"]
COMPLIANCE_STATUS = {"met", "partly", "not_met", "not_addressed"}
BOM_HEADER = ["item", "qty", "description", "manufacturer", "mpn", "supplier", "supplier_part", "url",
"unit_price_1", "unit_price_100", "unit_price_1000", "currency", "notes"]
HEADING_ID = re.compile(r"^#{1,6}\s+([A-Z][A-Z0-9]{1,7}-[A-Z0-9]{1,7}-\d{1,4})\b")
LOOSE_ID = re.compile(r"\b([A-Z][A-Z0-9]{1,7}-[A-Z]{2,7}-\d{2,4})\b")
FULL_DIRS = ["03-cad", "04-drawings", "05-calculations", "06-electronics", "07-firmware",
"08-safety-certification", "09-bom-cost", "10-prototype-test-plan", "11-renders"]
class Report:
def __init__(self):
self.errors, self.warnings, self.notes = [], [], []
def error(self, msg):
self.errors.append(msg)
def warn(self, msg):
self.warnings.append(msg)
def note(self, msg):
self.notes.append(msg)
def requirement_ids(path):
"""Ids of INPUT/requirements.md: headings first, loose tokens as a fallback."""
if not os.path.isfile(path):
return []
with open(path, encoding="utf-8", errors="replace") as fh:
text = fh.read()
ids = []
for line in text.splitlines():
m = HEADING_ID.match(line.strip())
if m and m.group(1) not in ids:
ids.append(m.group(1))
if not ids:
for m in LOOSE_ID.finditer(text):
if m.group(1) not in ids:
ids.append(m.group(1))
return ids
def nonempty_files(folder, exts=None):
out = []
for root, _dirs, files in os.walk(folder):
for name in files:
p = os.path.join(root, name)
if os.path.getsize(p) == 0:
continue
if exts and not name.lower().endswith(tuple(exts)):
continue
out.append(p)
return out
def split_paths(field):
parts = re.split(r"[;|\n]+|\s+(?=[\w./-]+/)", field or "")
return [p.strip().strip("`'\"") for p in parts if p.strip()]
def evidence_exists(deliv, ref):
ref = ref.split("#", 1)[0].split("::", 1)[0].strip()
if not ref:
return True
if ref.startswith("deliverables/"):
ref = ref[len("deliverables/"):]
ref = ref.lstrip("./")
m = re.match(r"^(.*?)(:\d+(-\d+)?)$", ref)
if m:
ref = m.group(1)
return os.path.exists(os.path.join(deliv, ref))
def check_compliance(deliv, ids, rep):
path = os.path.join(deliv, "01-requirements", "compliance.csv")
if not os.path.isfile(path):
rep.error("01-requirements/compliance.csv missing")
return
raw = open(path, encoding="utf-8-sig", errors="replace").read()
rows = list(csv.reader(io.StringIO(raw)))
if not rows:
rep.error("compliance.csv is empty")
return
header = [h.strip() for h in rows[0]]
if header != COMPLIANCE_HEADER:
rep.error(f"compliance.csv header is {header}, expected {COMPLIANCE_HEADER}")
return
seen = {}
for n, row in enumerate(rows[1:], start=2):
if not any(c.strip() for c in row):
continue
if len(row) != len(COMPLIANCE_HEADER):
rep.error(f"compliance.csv line {n}: {len(row)} fields, expected {len(COMPLIANCE_HEADER)} (quote fields with commas)")
continue
rid, status, _value, _unit, evidence, _note = [c.strip() for c in row]
if rid in seen:
rep.error(f"compliance.csv line {n}: duplicate id {rid} (first at line {seen[rid]})")
seen[rid] = n
if status not in COMPLIANCE_STATUS:
rep.error(f"compliance.csv line {n}: status '{status}' not in {sorted(COMPLIANCE_STATUS)}")
if status != "not_addressed" and not evidence:
rep.error(f"compliance.csv line {n} ({rid}): status {status} without evidence path")
for ref in split_paths(evidence):
if not evidence_exists(deliv, ref):
rep.error(f"compliance.csv line {n} ({rid}): evidence path not found: {ref}")
if ids:
missing = [i for i in ids if i not in seen]
extra = [i for i in seen if i not in ids]
if missing:
rep.error(f"compliance.csv lacks {len(missing)} requirement id(s): {', '.join(missing)}")
if extra:
rep.warn(f"compliance.csv has ids not in INPUT/requirements.md: {', '.join(extra)}")
rep.note(f"compliance.csv: {len(seen)} rows for {len(ids)} requirement ids")
else:
rep.warn("no requirement ids found in INPUT/requirements.md; row coverage not checked")
def check_ids_in_requirements(deliv, ids, rep):
folder = os.path.join(deliv, "01-requirements")
texts = ""
for p in nonempty_files(folder, (".md", ".yaml", ".yml", ".txt")):
texts += open(p, encoding="utf-8", errors="replace").read() + "\n"
if not texts:
rep.error("01-requirements/ has no requirements document (.md/.yaml) besides compliance.csv")
return
absent = [i for i in ids if i not in texts]
if absent:
rep.warn(f"requirement ids not mentioned in the 01-requirements documents: {', '.join(absent)}")
def num_or_null(x):
return x is None or isinstance(x, (int, float)) and not isinstance(x, bool)
def check_manifest(deliv, tier, rep):
path = os.path.join(deliv, "manifest.json")
if not os.path.isfile(path):
rep.error("manifest.json missing")
return
try:
man = json.load(open(path, encoding="utf-8"))
except Exception as exc: # noqa: BLE001
rep.error(f"manifest.json does not parse: {exc}")
return
if man.get("tier") != tier:
rep.error(f"manifest.json tier is {man.get('tier')!r}, expected {tier!r}")
files = man.get("files")
if not isinstance(files, list) or not files:
rep.error("manifest.json files[] missing or empty")
else:
listed = set()
for f in files:
if not isinstance(f, dict) or not {"path", "kind", "desc"} <= set(f):
rep.error(f"manifest.json files[] entry lacks path/kind/desc: {f}")
continue
listed.add(f["path"].lstrip("./"))
if not evidence_exists(deliv, f["path"]):
rep.error(f"manifest.json lists a file that does not exist: {f['path']}")
actual = set()
for p in nonempty_files(deliv):
rel = os.path.relpath(p, deliv)
if rel != "manifest.json":
actual.add(rel)
unlisted = sorted(a for a in actual if a not in listed and not any(a.startswith(l.rstrip("/") + "/") for l in listed))
if unlisted:
rep.warn(f"{len(unlisted)} deliverable file(s) not listed in manifest.json, e.g. {unlisted[:5]}")
kf = man.get("key_figures")
if not isinstance(kf, dict):
rep.error("manifest.json key_figures missing")
else:
for key in ("dimensions_mm", "mass_kg", "power_w", "unit_cost", "retail_price"):
if key not in kf:
rep.error(f"manifest.json key_figures.{key} missing (use null when unknown)")
dims = kf.get("dimensions_mm")
if dims is not None and not (isinstance(dims, list) and len(dims) == 3 and all(num_or_null(d) for d in dims)):
rep.error("key_figures.dimensions_mm must be [L, W, H] numbers or null")
for key in ("mass_kg", "power_w"):
if key in kf and not num_or_null(kf[key]):
rep.error(f"key_figures.{key} must be a number or null")
uc = kf.get("unit_cost")
if uc is not None:
if not isinstance(uc, dict) or not {"currency", "q1", "q100", "q1000"} <= set(uc):
rep.error("key_figures.unit_cost needs currency, q1, q100, q1000")
elif not all(num_or_null(uc[q]) for q in ("q1", "q100", "q1000")):
rep.error("key_figures.unit_cost q1/q100/q1000 must be numbers or null")
rp = kf.get("retail_price")
if rp is not None and (not isinstance(rp, dict) or not {"currency", "value"} <= set(rp) or not num_or_null(rp.get("value"))):
rep.error("key_figures.retail_price needs currency and a numeric value (or null)")
blocked = man.get("blocked")
if not isinstance(blocked, list):
rep.error("manifest.json blocked[] missing (use [] when nothing is blocked)")
else:
for b in blocked:
if not isinstance(b, dict) or not {"what", "why"} <= set(b):
rep.error(f"manifest.json blocked[] entry needs what and why: {b}")
def check_cost_json(deliv, rep, required):
path = os.path.join(deliv, "09-bom-cost", "cost.json")
if not os.path.isfile(path):
if required:
rep.error("09-bom-cost/cost.json missing")
return None
try:
cost = json.load(open(path, encoding="utf-8"))
except Exception as exc: # noqa: BLE001
rep.error(f"cost.json does not parse: {exc}")
return None
for key in ("currency", "bom", "unit_cost"):
if key not in cost:
rep.error(f"cost.json lacks {key}")
for key in ("bom", "unit_cost"):
block = cost.get(key)
if isinstance(block, dict) and not all(num_or_null(block.get(q)) for q in ("q1", "q100", "q1000")):
rep.error(f"cost.json {key} needs numeric q1, q100, q1000")
other = cost.get("other", {})
if not isinstance(other, dict):
rep.error("cost.json other must be an object of {name: {q1, q100, q1000}}")
else:
try:
for q in ("q1", "q100", "q1000"):
total = float(cost["bom"][q]) + sum(float(v.get(q) or 0) for v in other.values())
unit = float(cost["unit_cost"][q])
if unit > 0 and abs(total - unit) / unit > 0.02:
rep.warn(f"cost.json unit_cost.{q}={unit} differs from bom+other={total:.2f} by >2 %")
except (KeyError, TypeError, ValueError):
pass
return cost
def as_float(text):
try:
return float(str(text).replace(",", ".").strip())
except ValueError:
return None
def check_bom(deliv, cost, rep):
path = os.path.join(deliv, "09-bom-cost", "bom.csv")
if not os.path.isfile(path):
rep.error("09-bom-cost/bom.csv missing")
return
rows = list(csv.reader(io.StringIO(open(path, encoding="utf-8-sig", errors="replace").read())))
if not rows or [h.strip() for h in rows[0]] != BOM_HEADER:
rep.error(f"bom.csv header must be exactly {','.join(BOM_HEADER)}")
return
totals = {"unit_price_1": 0.0, "unit_price_100": 0.0, "unit_price_1000": 0.0}
currencies = set()
lines = 0
for n, row in enumerate(rows[1:], start=2):
if not any(c.strip() for c in row):
continue
if len(row) != len(BOM_HEADER):
rep.error(f"bom.csv line {n}: {len(row)} fields, expected {len(BOM_HEADER)}")
continue
rec = dict(zip(BOM_HEADER, [c.strip() for c in row]))
lines += 1
qty = as_float(rec["qty"])
if qty is None:
rep.error(f"bom.csv line {n}: qty '{rec['qty']}' is not a number")
continue
if not rec["mpn"] and not rec["supplier_part"]:
rep.warn(f"bom.csv line {n}: neither mpn nor supplier_part ({rec['description'][:40]})")
if rec["url"] and not rec["url"].startswith(("http://", "https://")):
rep.warn(f"bom.csv line {n}: url is not http(s)")
currencies.add(rec["currency"])
for col in totals:
price = as_float(rec[col])
if price is None:
rep.warn(f"bom.csv line {n}: {col} '{rec[col]}' is not a number")
else:
totals[col] += qty * price
rep.note(f"bom.csv: {lines} lines; sums 1/100/1000 = {totals['unit_price_1']:.2f} / "
f"{totals['unit_price_100']:.2f} / {totals['unit_price_1000']:.2f} {','.join(sorted(currencies))}")
if len(currencies) > 1:
rep.warn(f"bom.csv mixes currencies {sorted(currencies)}; totals not comparable")
elif cost and isinstance(cost.get("bom"), dict):
for q, col in (("q1", "unit_price_1"), ("q100", "unit_price_100"), ("q1000", "unit_price_1000")):
ref = as_float(cost["bom"].get(q))
if ref and abs(ref - totals[col]) / ref > 0.02:
rep.warn(f"cost.json bom.{q}={ref} differs from the bom.csv sum {totals[col]:.2f} by >2 %")
def toolchain_python():
"""The toolchain Python the RUN CONTRACT names (ddpy), else the venv, else this interpreter."""
for cand in (shutil.which("ddpy"), "/opt/dd/venv/bin/python"):
if cand and os.path.exists(cand):
return cand
return sys.executable
def check_scripts(folder, rep, run):
scripts = [p for p in nonempty_files(folder, (".py",))]
for script in scripts:
stem = os.path.splitext(script)[0]
results = stem + ".results.json"
if run:
try:
proc = subprocess.run([toolchain_python(), os.path.basename(script)], cwd=os.path.dirname(script),
capture_output=True, text=True, timeout=300)
if proc.returncode != 0:
rep.error(f"{os.path.relpath(script)} exits {proc.returncode}: {proc.stderr.strip()[-200:]}")
except subprocess.TimeoutExpired:
rep.error(f"{os.path.relpath(script)} runs longer than 5 minutes")
if not os.path.isfile(results):
rep.warn(f"{os.path.relpath(script)} has no {os.path.basename(results)}")
continue
try:
data = json.load(open(results, encoding="utf-8"))
bad = [k for k, v in data.items() if not (isinstance(v, dict) and "value" in v and "unit" in v)]
if bad:
rep.warn(f"{os.path.relpath(results)}: entries without value/unit: {bad[:5]}")
except Exception as exc: # noqa: BLE001
rep.error(f"{os.path.relpath(results)} does not parse: {exc}")
def check_full(deliv, rep, run):
for d in FULL_DIRS:
if not nonempty_files(os.path.join(deliv, d)):
rep.error(f"{d}/ missing or empty")
cad = os.path.join(deliv, "03-cad")
asm = os.path.join(cad, "assembly.step")
if not os.path.isfile(asm):
rep.error("03-cad/assembly.step missing")
else:
if not open(asm, "rb").read(64).lstrip().startswith(b"ISO-10303-21"):
rep.error("03-cad/assembly.step is not a STEP file (no ISO-10303-21 header)")
steps = [p for p in nonempty_files(cad, (".step", ".stp")) if not p.endswith("assembly.step")]
if not steps:
rep.error("03-cad/ has no per-part STEP files")
if not nonempty_files(cad, (".py", ".fcstd")):
rep.error("03-cad/ has no parametric source (.py or .FCStd)")
if not nonempty_files(os.path.join(deliv, "04-drawings"), (".pdf", ".svg")):
rep.error("04-drawings/ has no PDF or SVG drawing")
check_scripts(os.path.join(deliv, "05-calculations"), rep, run)
ele = os.path.join(deliv, "06-electronics")
if not nonempty_files(ele, (".kicad_pro",)) or not nonempty_files(ele, (".kicad_sch",)):
rep.error("06-electronics/ lacks a KiCad project (.kicad_pro + .kicad_sch)")
names = " ".join(os.path.basename(p).lower() for p in nonempty_files(ele))
if "erc" not in names:
rep.warn("06-electronics/ has no file named like an ERC report")
if nonempty_files(ele, (".kicad_pcb",)) and "drc" not in names:
rep.warn("06-electronics/ has a board but no file named like a DRC report")
if not os.path.isfile(os.path.join(deliv, "07-firmware", "platformio.ini")):
rep.error("07-firmware/platformio.ini missing")
cost = check_cost_json(deliv, rep, required=True)
check_bom(deliv, cost, rep)
if not os.path.isfile(os.path.join(deliv, "09-bom-cost", "cost.md")):
rep.error("09-bom-cost/cost.md missing")
ren = os.path.join(deliv, "11-renders")
if not nonempty_files(ren, (".png",)):
rep.error("11-renders/ has no PNG render")
glb = os.path.join(ren, "assembly.glb")
if not os.path.isfile(glb):
rep.error("11-renders/assembly.glb missing")
elif open(glb, "rb").read(4) != b"glTF":
rep.error("11-renders/assembly.glb is not a binary glTF file")
def main():
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--tier", choices=["concept", "full"], required=True)
ap.add_argument("--run-dir", default=".", help="run directory (default: current directory)")
ap.add_argument("--run-scripts", action="store_true", help="execute calculation scripts (≤5 min each)")
ap.add_argument("--json", action="store_true", help="print a JSON summary as the last line")
args = ap.parse_args()
run_dir = os.path.abspath(args.run_dir)
deliv = os.path.join(run_dir, "deliverables")
rep = Report()
if not os.path.isdir(deliv):
rep.error("deliverables/ missing")
else:
ids = requirement_ids(os.path.join(run_dir, "INPUT", "requirements.md"))
if not os.path.isdir(os.path.join(deliv, "01-requirements")):
rep.error("01-requirements/ missing")
else:
check_compliance(deliv, ids, rep)
check_ids_in_requirements(deliv, ids, rep)
if not nonempty_files(os.path.join(deliv, "02-concept"), (".md",)):
rep.error("02-concept/ has no markdown document")
report = os.path.join(deliv, "REPORT.md")
if not os.path.isfile(report):
rep.error("REPORT.md missing")
elif os.path.getsize(report) < 1500:
rep.warn("REPORT.md is shorter than 1500 bytes")
check_manifest(deliv, args.tier, rep)
check_scripts(os.path.join(deliv, "02-concept"), rep, args.run_scripts)
if args.tier == "concept":
check_cost_json(deliv, rep, required=False)
started = [d for d in FULL_DIRS if d != "09-bom-cost" and nonempty_files(os.path.join(deliv, d))]
if started:
rep.warn(f"concept tier: detailed-design folders present ({', '.join(started)}); they are not judged at this tier")
else:
check_full(deliv, rep, args.run_scripts)
for msg in rep.notes:
print(f"NOTE {msg}")
for msg in rep.warnings:
print(f"WARN {msg}")
for msg in rep.errors:
print(f"ERROR {msg}")
print(f"dd_check: tier={args.tier} errors={len(rep.errors)} warnings={len(rep.warnings)} -> "
f"{'PASS' if not rep.errors else 'FAIL'}")
if args.json:
print(json.dumps({"tier": args.tier, "pass": not rep.errors, "errors": rep.errors,
"warnings": rep.warnings, "notes": rep.notes}))
return 0 if not rep.errors else 1
if __name__ == "__main__":
sys.exit(main())
files/method/gates/meshlib.py (518 wierszy)
#!/usr/bin/env python3
"""meshlib.py - shared code of post.py and board_check.py (not run directly). Standard library + PyYAML
(re-executes under the toolchain Python `ddpy` if this interpreter lacks it); pint converts units when present."""
import datetime as dt
import fcntl
import fnmatch
import glob
import hashlib
import json
import os
import re
import shutil
import sys
try:
import yaml
except ImportError: # pragma: no cover - depends on the interpreter
_ALT = shutil.which("ddpy") or "/opt/dd/venv/bin/python"
if os.environ.get("MESH_REEXEC") != "1" and os.path.exists(_ALT):
os.environ["MESH_REEXEC"] = "1"
os.execv(_ALT, [_ALT, os.path.abspath(sys.argv[0])] + sys.argv[1:])
yaml = None
RUN = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
BOARD, DELIV = os.path.join(RUN, "board"), os.path.join(RUN, "deliverables")
LOG = os.path.join(BOARD, "log.jsonl")
ROLE_RE = re.compile(r"^[A-Z][A-Z0-9]{1,7}$")
LABELS = {"SOURCE", "CALCULATION", "SIMULATION", "MEASUREMENT", "ASSUMPTION"}
TYPES = {"Q", "CR", "ANS", "DECISION", "OBJECTION", "CALL", "VOTE", "VETO", "LIFT", "INFO", "FYI", "DONE", "BLOCKED",
"PHASE", "NUDGE", "TALLY", "CLOSE"}
CONVENER_TYPES = {"PHASE", "NUDGE", "TALLY", "CLOSE", "INFO"}
ASK_TYPES = {"Q", "CR", "ANS", "DECISION", "OBJECTION", "CALL", "VETO", "LIFT", "INFO", "PHASE", "NUDGE", "TALLY", "CLOSE"}
NOT_TO_CONVENER = {"Q", "CR", "OBJECTION", "VETO"} # the convener decides nothing, so it can never close these
CLOSERS = {"Q": {"ANS", "DECISION"}, "CR": {"ANS", "DECISION"}, "OBJECTION": {"ANS", "DECISION"}, "VETO": {"LIFT"}, "CALL": {"TALLY"}}
STALL_MIN = 15
PLANS = {"concept": [("SETUP", 10), ("FRAME", 28), ("CONVERGE", 43), ("WRITE", 58), ("CLOSE", 63)],
"full": [("SETUP", 10), ("FRAME", 30), ("CONVERGE", 50), ("WRITE", 75), ("BUILD", 330), ("CROSSCHECK", 420),
("FINISH", 510), ("CLOSE", 540)]}
CLOSE_BY = {"concept": 80, "full": 570}
HEAD_ID = re.compile(r"^#{1,6}\s+([A-Z][A-Z0-9]{1,7}-[A-Z0-9]{1,7}-\d{1,4})\b")
LOOSE_ID = re.compile(r"\b([A-Z][A-Z0-9]{1,7}-[A-Z]{2,7}-\d{2,4})\b")
# a requirement's weight is a FIELD line ("- Priority: 4/5", "**Weight:** 3", "| Priority | 4 |"), never prose such as
# "the unit's weight: 2.5 kg"; Priority wins over Weight, and a Weight field must not carry a unit
_FIELD = r"^\s*(?:[-*+]|\d+[.)])?\s*\**\s*{0}\s*\**\s*[:=]\s*\**\s*(\d+(?:\.\d+)?)\s*\**\s*{1}"
_CELL = r"^\s*\|\s*\**\s*{0}\s*\**\s*\|\s*\**\s*(\d+(?:\.\d+)?)\s*(?:/\s*\d+(?:\.\d+)?)?\s*\**\s*\|"
PRIO_FIELDS = [(re.compile(_FIELD.format("priority", r"(?:[/(.,;]|$|\s)"), re.I), re.compile(_CELL.format("priority"), re.I)),
(re.compile(_FIELD.format("weight", r"(?:/\s*\d+(?:\.\d+)?)?\s*(?:\(.*)?$"), re.I), re.compile(_CELL.format("weight"), re.I))]
def rel(path):
return os.path.relpath(path, RUN)
def now():
return dt.datetime.now().astimezone()
def iso(t):
return t.isoformat(timespec="seconds")
def parse_t(text):
try:
t = dt.datetime.fromisoformat(str(text))
return t if t.tzinfo else t.astimezone()
except ValueError:
return None
def hm(t):
return t.strftime("%H:%M") if t else "?"
def minutes(a, b):
return (b - a).total_seconds() / 60.0
def sha(path):
return hashlib.sha256(open(path, "rb").read()).hexdigest()[:12]
def load_yaml(path):
text = open(path, encoding="utf-8").read()
return yaml.safe_load(text) if yaml else json.loads(text)
def num(x):
return isinstance(x, (int, float)) and not isinstance(x, bool)
def vals(v):
return list(v) if isinstance(v, (list, tuple)) else [v]
# ---------- roster and ownership ----------
def roster():
"""{ROLE: {session, status, brief, owns[]}} from the table rows of board/ROSTER.md."""
out, path = {}, os.path.join(BOARD, "ROSTER.md")
for line in open(path, encoding="utf-8") if os.path.isfile(path) else []:
c = [x.strip() for x in line.strip().strip("|").split("|")]
if len(c) >= 5 and ROLE_RE.match(c[0]):
out[c[0]] = {"session": c[1], "status": c[2].lower(), "brief": c[3],
"owns": [p.strip().strip("`") for p in c[4].split(",") if p.strip()]}
return out
def patterns(role, info):
"""The ROSTER row's paths + the implicit board and calculation paths of the role and of every role merged
into it (its brief cell then reads 'method/roles/REQ.md + method/roles/COST.md')."""
pats = list(info.get("owns", []))
if role == "CONVENER":
return pats
for c in [role] + [x for x in re.findall(r"roles/([A-Z][A-Z0-9]{1,7})\.md", info.get("brief", "")) if x not in (role, "DOMAIN")]:
pats += [f"board/facts/{c}.yaml", f"board/ideas/{c}.md", f"board/decisions/D-{c}-*", f"board/votes/*/{c}.yaml",
f"board/votes/{c}-*/BALLOT.yaml", f"board/out/{c}-*", f"deliverables/02-concept/sections/{c}.md",
f"deliverables/02-concept/calc/{c.lower()}_*", f"deliverables/05-calculations/{c.lower()}_*"]
return pats
def hosts(ros=None):
"""{code: roster role}: a roster role maps to itself, a merged role code (its brief named in another role's brief
cell, e.g. COST in 'method/roles/REQ.md + method/roles/COST.md') to the role that holds it."""
ros = roster() if ros is None else ros
out = {r: r for r in ros}
for r, info in ros.items():
for c in re.findall(r"roles/([A-Z][A-Z0-9]{1,7})\.md", info.get("brief", "")):
out.setdefault(c, r)
out.pop("DOMAIN", None)
return out
def owner_of(path, ros):
"""(owner or None, [candidates]); the longest matching pattern wins, a tie is ambiguous."""
best, who = -1, []
for role, info in ros.items():
for pat in patterns(role, info):
if fnmatch.fnmatchcase(path, pat):
if len(pat) > best:
best, who = len(pat), [role]
elif len(pat) == best and role not in who:
who.append(role)
return (who[0] if len(who) == 1 else None), who
def live_session(info):
s = (info or {}).get("session", "")
return s if s not in ("", "-", "PENDING") and (info or {}).get("status") in ("live", "done") else None
# ---------- log ----------
def read_log():
out = []
for n, line in enumerate(open(LOG, encoding="utf-8") if os.path.isfile(LOG) else [], 1):
if line.strip():
try:
m = json.loads(line)
m.setdefault("to", [])
out.append(m)
except ValueError:
out.append({"id": f"BAD-{n}", "type": "BAD", "from": "?", "to": [], "text": line[:80], "ts": ""})
return out
def idn(m):
try:
return int(str(m.get("id", "")).split("-")[1])
except (IndexError, ValueError):
return 0
def append_log(entry):
os.makedirs(BOARD, exist_ok=True)
with open(os.path.join(BOARD, ".log.lock"), "w") as lock:
fcntl.flock(lock, fcntl.LOCK_EX)
entry = {"id": f"M-{max([idn(m) for m in read_log()] or [0]) + 1:04d}", "ts": iso(now()), **entry}
with open(LOG, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
return entry
def addressed(m, role):
return role in m.get("to", []) or ("ALL" in m.get("to", []) and m.get("from") != role)
def open_threads(log, t=None):
"""[(msg, age_min, pending)] of Q/CR/OBJECTION/VETO/CALL messages still awaiting a closing reply. pending = the
roles that must act (named addressees must each reply; a message to ALL is closed by the first reply; a VETO
waits for its addressees to redesign or mitigate until SAFE's LIFT closes it; a CALL waits for the TALLY).
The convener is never pending on a question: it decides nothing (post.py refuses such messages)."""
t, replies, out = t or now(), {}, []
for m in log:
replies.setdefault(m.get("re"), []).append(m)
for m in log:
closers = CLOSERS.get(m.get("type"))
if not closers:
continue
who = {r.get("from") for r in replies.get(m["id"], []) if r.get("type") in closers}
if m["type"] == "VETO":
pending = [] if m.get("from") in who else ([r for r in m.get("to", []) if r not in ("ALL", "CONVENER")] or ["ALL"])
elif m["type"] == "CALL":
pending = [] if who else ["CONVENER"]
elif "ALL" in m.get("to", []):
pending = [] if who else ["ALL"]
else:
pending = [r for r in m.get("to", []) if r not in who and r != "CONVENER"]
if pending:
ts = parse_t(m.get("ts", ""))
out.append((m, minutes(ts, t) if ts else 0.0, pending))
return out
def pending_for(pending, m, role):
return role in pending or ("ALL" in pending and m.get("from") != role)
def ask_lines(entry, ros):
"""Exact mcp__oivo__sessions ask payloads for a logged entry (lines to print)."""
typ, frm = entry["type"], entry["from"]
if typ not in ASK_TYPES | {"FYI"}:
return ["no ask needed: board only"]
targets = [r for r in ros if r not in ("CONVENER", frm)] if "ALL" in entry["to"] else list(entry["to"])
refs = [r["path"] if isinstance(r, dict) else r for r in entry.get("refs", [])]
msg = (f"[{entry['id']} {typ} {frm}->{','.join(entry['to'])}" + (f" re {entry['re']}" if entry.get("re") else "") + "] "
+ entry["text"] + (f" | refs: {', '.join(refs)}" if refs else ""))
if typ in ("Q", "CR", "OBJECTION"):
msg += (f" | answer by {hm(now() + dt.timedelta(minutes=STALL_MIN))} with: python3 method/gates/post.py --from <YOU> "
f"--to {frm} --type DECISION|ANS --re {entry['id']} --text \"...\"")
elif typ == "CALL":
msg += f" | vote: write your vote file, then post.py --type VOTE --re {entry['id']} (board only)"
out = []
for r in targets:
sid = live_session(ros.get(r))
if r == "CONVENER":
out.append("CONVENER: no ask (the convener reads the board)")
elif not sid:
out.append(f"{r}: no live session ({(ros.get(r) or {}).get('status', 'unknown role')}): logged only; it sees this in "
f"`post.py inbox`; re-print later with: python3 method/gates/post.py show {entry['id']}")
else:
# redirect: a busy peer reads it at its next tool boundary, an idle one is woken; fyi (FYI only): a copy shown
# at the recipient's next turn that never wakes it - for nothing anybody must act on soon
extra = {"fyi": True} if typ == "FYI" else {"redirect": True}
out.append("ASK " + json.dumps({"action": "ask", "session": sid, "message": msg, **extra}, ensure_ascii=False))
return out
# ---------- clock ----------
def clock():
path = os.path.join(BOARD, "CLOCK.json")
return json.load(open(path, encoding="utf-8")) if os.path.isfile(path) else None
def phase_now(ck, t=None):
"""(name, end, next_name, next_end); OVERTIME after the last phase."""
t, ph = t or now(), [(p["name"], parse_t(p["end"])) for p in ck["phases"]]
for i, (name, end) in enumerate(ph):
if t < end:
nxt = ph[i + 1] if i + 1 < len(ph) else (None, None)
return name, end, nxt[0], nxt[1]
return "OVERTIME", None, None, None
def clock_line(ck=None):
ck = ck or clock()
if not ck:
return "CLOCK not initialised (board_check.py init)"
t = now()
name, end, nxt, nend = phase_now(ck, t)
cb = parse_t(ck.get("close_by", ""))
return (f"CLOCK {hm(t)} {ck['tier']} phase {name}" + (f" until {hm(end)} ({minutes(t, end):.0f} min left)" if end else "")
+ (f" | next {nxt} until {hm(nend)}" if nxt else "") + (f" | close by {hm(cb)} ({minutes(t, cb):.0f} min)" if cb else ""))
# ---------- requirements ----------
def requirements():
"""(ids in order, {id: weight}) from INPUT/requirements.md: heading ids first, loose ids as a fallback. The weight
is the requirement block's Priority field line, else its Weight field line; ids without one are left out of the
dict (callers warn and use weight 1)."""
path = os.path.join(RUN, "INPUT", "requirements.md")
text = open(path, encoding="utf-8", errors="replace").read() if os.path.isfile(path) else ""
ids, found, cur = [], {}, None
for line in text.splitlines():
m = HEAD_ID.match(line.strip())
if m:
cur = m.group(1)
ids += [] if cur in ids else [cur]
elif line.lstrip().startswith("#"):
cur = None
elif cur:
for rank, pats in enumerate(PRIO_FIELDS):
hit = next((p.match(line) for p in pats if p.match(line)), None)
if hit and rank not in found.setdefault(cur, {}):
found[cur][rank] = float(hit.group(1))
weights = {rid: by[min(by)] for rid, by in found.items() if by}
return ids or list(dict.fromkeys(LOOSE_ID.findall(text))), weights
def reqmap(problems=None):
"""{requirement id: [raters]} from board/reqmap.yaml (`ID: [ROLE]` lines). A list of one-key mappings is read as
one mapping; any other shape gives {} and a line in `problems` (when a list is passed)."""
path, data = os.path.join(BOARD, "reqmap.yaml"), {}
note = (problems.append if isinstance(problems, list) else (lambda _x: None))
try:
data = load_yaml(path) if os.path.isfile(path) else {}
except Exception as exc: # noqa: BLE001
note(f"board/reqmap.yaml does not parse: {str(exc).splitlines()[0][:120]}")
return {}
if isinstance(data, list) and all(isinstance(e, dict) for e in data):
merged = {}
for e in data:
merged.update(e)
data = merged
if data and not isinstance(data, dict):
note("board/reqmap.yaml must be a mapping of `REQ-ID: [ROLE, ...]` lines")
return {}
out = {}
for k, v in (data or {}).items():
if isinstance(v, dict) or v in (None, "", []):
note(f"board/reqmap.yaml: {k} needs a role or a list of roles, got {v!r}")
continue
v = re.split(r"[,\s]+", v) if isinstance(v, str) else vals(v)
out[str(k).strip()] = [str(x).strip().upper() for x in v if str(x).strip()]
return out
# ---------- units ----------
_T = {"m": (1, "m"), "mm": (1e-3, "m"), "cm": (1e-2, "m"), "km": (1e3, "m"), "um": (1e-6, "m"), "kg": (1, "kg"), "g": (1e-3, "kg"),
"mg": (1e-6, "kg"), "t": (1e3, "kg"), "W": (1, "W"), "mW": (1e-3, "W"), "kW": (1e3, "W"), "J": (1, "J"), "kJ": (1e3, "J"),
"MJ": (1e6, "J"), "Wh": (3600, "J"), "kWh": (3.6e6, "J"), "s": (1, "s"), "ms": (1e-3, "s"), "min": (60, "s"), "h": (3600, "s"),
"l": (1e-3, "m3"), "L": (1e-3, "m3"), "ml": (1e-6, "m3"), "mL": (1e-6, "m3"), "m3": (1, "m3"), "cm3": (1e-6, "m3"),
"mm3": (1e-9, "m3"), "m2": (1, "m2"), "cm2": (1e-4, "m2"), "mm2": (1e-6, "m2"), "V": (1, "V"), "mV": (1e-3, "V"),
"kV": (1e3, "V"), "A": (1, "A"), "mA": (1e-3, "A"), "uA": (1e-6, "A"), "Pa": (1, "Pa"), "kPa": (1e3, "Pa"),
"MPa": (1e6, "Pa"), "bar": (1e5, "Pa"), "mbar": (100, "Pa"), "N": (1, "N"), "kN": (1e3, "N"), "Hz": (1, "Hz"),
"kHz": (1e3, "Hz"), "MHz": (1e6, "Hz"), "GHz": (1e9, "Hz"), "K": (1, "K"), "ohm": (1, "ohm"), "kohm": (1e3, "ohm"),
"%": (0.01, "1"), "1": (1, "1"), "": (1, "1")}
_UREG = []
def _pint(value, unit):
if not _UREG:
try:
import pint
_UREG.append(pint.UnitRegistry())
except Exception: # noqa: BLE001
_UREG.append(None)
try:
q = _UREG[0].Quantity(value, unit).to_base_units()
return float(q.magnitude), str(q.units)
except Exception: # noqa: BLE001 - no pint, or a unit pint does not know
return None
def to_base(value, unit):
"""(value in base units, dimension key); an unknown unit (EUR ...) is its own dimension."""
unit = str("1" if unit is None else unit).strip()
if unit in _T:
return value * _T[unit][0], _T[unit][1]
a, _, b = (x.strip() for x in unit.partition("/"))
if b and a in _T and b in _T:
return value * _T[a][0] / _T[b][0], f"{_T[a][1]}/{_T[b][1]}"
return (None if re.fullmatch(r"[A-Z]{3}", unit) else _pint(value, unit)) or (value, unit)
def compare(a_val, a_unit, b_val, b_unit):
"""[(a, b)] in common base units, or None when not comparable."""
av, bv = vals(a_val), vals(b_val)
if len(av) != len(bv) or not all(num(x) for x in av + bv):
return None
pairs = [(to_base(x, a_unit), to_base(y, b_unit)) for x, y in zip(av, bv)]
return None if any(p[0][1] != p[1][1] for p in pairs) else [(p[0][0], p[1][0]) for p in pairs]
def close_enough(a, b, rtol=0.01):
return abs(a - b) <= rtol * max(abs(a), abs(b), 1e-12)
# ---------- facts ----------
def facts_check():
"""(facts {id: fact + owner}, problems [(severity, role, text)], budgets [(name, total, unit, limit, n)])."""
facts, uses, probs, budgets = {}, [], [], []
for path in sorted(glob.glob(os.path.join(BOARD, "facts", "*.y*ml"))):
role = os.path.splitext(os.path.basename(path))[0]
try:
data = load_yaml(path) or {}
except Exception as exc: # noqa: BLE001
probs.append(("ERROR", role, f"{rel(path)} does not parse: {str(exc).splitlines()[0][:160]}"))
continue
if not isinstance(data, dict):
probs.append(("ERROR", role, f"{rel(path)} must be a mapping with owner, facts, uses"))
continue
if data.get("owner") != role:
probs.append(("ERROR", role, f"{rel(path)}: owner is {data.get('owner')!r}, expected {role}"))
for f in data.get("facts") or []:
if not isinstance(f, dict):
probs.append(("ERROR", role, f"{rel(path)}: fact entry is not a mapping: {f!r}"[:200]))
continue
fid = str(f.get("id", ""))
miss = [k for k in ("value", "unit", "label", "source") if f.get(k) in (None, "")]
for bad, text in ((not fid.startswith(role.lower() + "."), f"fact id {fid!r} must start with '{role.lower()}.' (owner prefix)"),
(fid in facts, f"duplicate fact id {fid}"), (miss, f"{fid}: missing {', '.join(miss)}"),
(f.get("label") not in LABELS, f"{fid}: label {f.get('label')!r} not in {sorted(LABELS)}"),
(f.get("value") is not None and not all(num(x) for x in vals(f.get("value"))),
f"{fid}: value {f.get('value')!r} is not a number or a list of numbers")):
if bad:
probs.append(("ERROR", role, text))
facts[fid] = dict(f, owner=role)
uses += [(role, u) for u in data.get("uses") or [] if isinstance(u, dict)]
for role, u in uses:
f = facts.get(str(u.get("id")))
pairs = compare(u.get("value"), u.get("unit"), f.get("value"), f.get("unit")) if f else None
if not f:
probs.append(("WARN", role, f"uses unknown fact {u.get('id')}"))
elif pairs is None:
probs.append(("WARN", role, f"uses {u.get('id')}: {u.get('value')} {u.get('unit')} not comparable with the owner's {f.get('value')} {f.get('unit')}"))
elif not all(close_enough(a, b) for a, b in pairs):
probs.append(("ERROR", role, f"STALE: uses {u.get('id')}={u.get('value')} {u.get('unit')} but {f['owner']} now has {f.get('value')} {f.get('unit')}"))
for fid, f in facts.items():
for op in ("le", "ge"):
o = facts.get(str(f.get(op))) if f.get(op) else None
pairs = compare(f.get("value"), f.get("unit"), o.get("value"), o.get("unit")) if o else None
if f.get(op) and not o:
probs.append(("WARN", f["owner"], f"{fid} {op} {f[op]}: no such fact"))
elif f.get(op) and pairs is None:
probs.append(("WARN", f["owner"], f"{fid} {op} {f[op]}: units not comparable"))
elif pairs and not all((a <= b * 1.000001) if op == "le" else (a >= b * 0.999999) for a, b in pairs):
probs.append(("ERROR", f["owner"], f"CONSTRAINT {fid}={f.get('value')} {f.get('unit')} must be {'<=' if op == 'le' else '>='} "
f"{f[op]}={o.get('value')} {o.get('unit')} ({o['owner']})"))
groups = {}
for f in facts.values():
if f.get("budget"):
groups.setdefault(str(f["budget"]), []).append(f)
for name, items in sorted(groups.items()):
lim = facts.get(f"arch.budget.{name}")
unit = (lim or items[0]).get("unit")
ref, total, ok = to_base(1.0, unit), 0.0, True
for f in items:
b = to_base(f["value"], f.get("unit")) if num(f.get("value")) else (None, None)
if b[1] != ref[1]:
ok = False
probs.append(("WARN", f["owner"], f"budget {name}: {f.get('id')} ({f.get('value')} {f.get('unit')}) not comparable with {unit}"))
else:
total += b[0] / ref[0]
budgets.append((name, round(total, 4), unit, lim.get("value") if lim else None, len(items)))
if lim and ok and num(lim.get("value")) and total > lim["value"] * 1.000001:
probs.append(("ERROR", "ARCH", f"BUDGET {name}: contributions sum to {total:.4g} {unit} > arch.budget.{name} = {lim['value']} {unit}"))
host = hosts() # findings in a merged role's files (board/facts/COST.yaml) go to the role that holds it
return facts, [(s, host.get(r, r), x) for s, r, x in probs], budgets
# ---------- votes ----------
def load_ballot(topic):
"""(ballot mapping or None, problem or None) of board/votes/<topic>/BALLOT.yaml."""
path = os.path.join(BOARD, "votes", str(topic), "BALLOT.yaml")
if not os.path.isfile(path):
return None, "missing"
try:
b = load_yaml(path)
except Exception as exc: # noqa: BLE001
return None, f"does not parse: {str(exc).splitlines()[0][:120]}"
if not isinstance(b, dict):
return None, "is not a mapping (method/templates/ballot.yaml)"
return b, None
def ballot_voters(b, ros=None):
"""Voter roles of a ballot: `voters: ALL` = every peer of the roster; merged codes count as the role holding them;
roles not in the roster are left out."""
if not isinstance(b, dict):
return []
ros = roster() if ros is None else ros
v = b.get("voters", "ALL")
v = re.split(r"[,\s]+", v) if isinstance(v, str) else v if isinstance(v, list) else []
v = [str(x).strip().upper() for x in v if str(x).strip()]
peers, host, out = [r for r in ros if r != "CONVENER"], hosts(ros), []
for x in (peers if "ALL" in v else v):
h = host.get(x, x)
if h in peers and h not in out:
out.append(h)
return out
def criteria(ballot, issues=None):
"""{criterion: {weight, roles, req}}; requirement weights come from INPUT, never from the ballot."""
note = issues.append if isinstance(issues, list) else (lambda _x: None)
ids, weights = requirements()
host, probs = hosts(), []
rmap, sel = reqmap(probs), ballot.get("requirements", "none")
for p in probs:
note(p)
chosen = ids if str(sel).strip().lower() == "all" else [str(x) for x in sel] if isinstance(sel, list) else []
crit = {rid: {"weight": weights.get(rid, 1.0), "roles": [host.get(r, r) for r in rmap.get(rid) or ["REQ"]], "req": True}
for rid in chosen}
if chosen and [rid for rid in chosen if rid not in weights]:
note(f"no Priority/Weight field found in INPUT/requirements.md for {', '.join(r for r in chosen if r not in weights)}: weight 1 used")
total = sum(c["weight"] for c in crit.values())
raw = ballot.get("extra_criteria") or {}
if isinstance(raw, list) and all(isinstance(e, dict) for e in raw):
raw = {k: v for e in raw for k, v in e.items()}
if not isinstance(raw, dict):
note("BALLOT.yaml extra_criteria is not a mapping: ignored")
raw = {}
extra = {}
for k, v in raw.items():
spec = v if isinstance(v, dict) else {"weight": v}
try:
spec = dict(spec, **{x: float(spec[x]) for x in ("share", "weight") if x in spec})
except (TypeError, ValueError):
note(f"BALLOT.yaml extra criterion {k}: share or weight is not a number: ignored")
continue
extra[str(k)] = spec
shares = sum(v.get("share", 0) for v in extra.values())
for cid, spec in extra.items():
roles = spec.get("roles", "ALL")
roles = re.split(r"[,\s]+", roles) if isinstance(roles, str) else vals(roles)
roles = [str(r).strip().upper() for r in roles if str(r).strip()]
w = (spec["share"] * (total / (1 - shares) if total > 0 and shares < 1 else 10)) if "share" in spec else spec.get("weight", 1.0)
crit[cid] = {"weight": w, "roles": ["ALL"] if "ALL" in roles or not roles else [host.get(r, r) for r in roles], "req": False}
return crit
files/method/gates/post.py (193 wierszy)
#!/usr/bin/env python3
"""post.py - the only way to write board/log.jsonl (append-only, flock-protected) and to read your inbox.
Log a message, then send every printed ASK line with mcp__oivo__sessions (same fields):
python3 method/gates/post.py --from MECH --to ELEC --type CR [--re M-0007] [--ref PATH ...] --text "..."
(--to takes ROLE, ROLE,ROLE or ALL; text <= 600 characters: put details in a file and --ref it)
Other commands:
python3 method/gates/post.py inbox --role MECH [--all] what you must answer or vote on, what is new for you since
your last inbox (--all: everything addressed to you), your open requests
python3 method/gates/post.py show M-0042 re-print the ASK lines of a logged message
python3 method/gates/post.py clock current phase, time left, close-by time
Types: Q CR ANS DECISION OBJECTION CALL VOTE VETO LIFT INFO FYI DONE BLOCKED (peers; VETO/LIFT only SAFE)
PHASE NUDGE TALLY CLOSE INFO (convener). --re is required for ANS, LIFT and TALLY. Every ASK line carries
redirect: true except FYI (fyi: true: shown at the recipient's next turn, never wakes it).
"""
import json
import argparse
import os
import sys
sys.dont_write_bytecode = True
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import meshlib as M # noqa: E402
def fail(msg):
print(f"REFUSED: {msg}")
sys.exit(2)
def cmd_post(a):
ros = M.roster()
if not ros:
fail("board/ROSTER.md missing: the convener has not initialised the board yet")
host, notes = M.hosts(ros), []
frm = host.get(a.sender.upper(), a.sender.upper())
to = list(dict.fromkeys(host.get(x, x) for x in (t.strip().upper() for t in a.to.split(",")) if x))
notes += [f"NOTE {x} is held by {host[x]} (merged): addressed to {host[x]}"
for x in (y.strip() for y in a.to.upper().split(",")) if host.get(x, x) != x]
typ = a.type.upper()
if frm not in ros:
fail(f"--from {frm} is not a role in board/ROSTER.md ({', '.join(ros)})")
bad = [t for t in to if t != "ALL" and t not in ros]
if bad or not to:
fail(f"--to {a.to}: unknown role(s) {bad}, or not listed yet (the convener adds the roles of the second wave "
f"minutes after the first); use roles of board/ROSTER.md or ALL, and meanwhile go on on a logged ASSUMPTION")
if typ not in M.TYPES:
fail(f"--type {typ} not in {sorted(M.TYPES)}")
if frm == "CONVENER" and typ not in M.CONVENER_TYPES:
fail(f"the convener never decides, votes or requests changes; allowed: {sorted(M.CONVENER_TYPES)}")
if frm != "CONVENER" and typ in M.CONVENER_TYPES - {"INFO"}:
fail(f"{typ} is a convener message")
if typ in ("VETO", "LIFT") and frm != "SAFE":
fail("only SAFE vetoes or lifts a veto; send an OBJECTION instead")
if typ in M.NOT_TO_CONVENER and "CONVENER" in to:
fail(f"{typ} --to CONVENER: the convener decides nothing and cannot close it. Design questions and change "
f"requests go to the owner (board/ROSTER.md; `python3 method/gates/board_check.py owner <path>`), disputes "
f"follow method/PROTOCOL.md section 7; a procedural problem (script, roster, clock, a silent peer) is an "
f"INFO --to CONVENER")
if typ in ("ANS", "LIFT", "TALLY") and not a.re:
fail(f"{typ} needs --re <message id it answers>")
log = M.read_log()
if a.re and a.re not in {m["id"] for m in log}:
fail(f"--re {a.re}: no such message in board/log.jsonl")
text = " ".join(a.text.split())
if not text or len(text) > 600:
fail(f"text has {len(text)} characters; 1..600 allowed (write details into a file you own and --ref it)")
refs = []
for i, r in enumerate(a.ref or []):
p = os.path.join(M.RUN, r)
refs.append({"path": r, "sha": M.sha(p)} if os.path.isfile(p) else r)
if typ == "DECISION" and i == 0 and r.startswith(("deliverables/", "board/")):
owner, _ = M.owner_of(r, ros)
if owner and owner != frm:
notes.append(f"WARN {r} is owned by {owner}: only the owner decides about it (send a CR)")
if typ == "CALL" and not any(isinstance(r, dict) and r["path"].endswith("BALLOT.yaml") for r in refs):
fail("CALL needs --ref board/votes/<CALLER>-<topic>/BALLOT.yaml (existing file)")
entry = {"from": frm, "to": to, "type": typ, "text": text}
if a.re:
entry["re"] = a.re
if refs:
entry["refs"] = refs
entry = M.append_log(entry)
print(f"logged {entry['id']} {typ} {frm}->{','.join(to)} at {M.hm(M.parse_t(entry['ts']))}")
for n in notes:
print(n)
for line in M.ask_lines(entry, ros):
print(line)
def cmd_show(a):
for m in M.read_log():
if m.get("id") == a.id:
print(f"{m['id']} {m['ts']} {m['type']} {m['from']}->{','.join(m['to'])}: {m['text']}")
for line in M.ask_lines(m, M.roster()):
print(line)
return
fail(f"no message {a.id}")
def short(m, t):
ts = M.parse_t(m.get("ts", ""))
age = f"{M.minutes(ts, t):.0f} min" if ts else "?"
return f"{m['id']} {M.hm(ts)} {m['type']} {m['from']}->{','.join(m['to'])} ({age}): {m['text'][:220]}"
def cmd_inbox(a):
role, t = a.role.upper(), M.now()
ros, log = M.roster(), M.read_log()
if role not in ros:
fail(f"{role} is not in board/ROSTER.md")
print(M.clock_line())
phase_msg = [m for m in log if m["type"] in ("PHASE", "CLOSE")]
if phase_msg:
print("PHASE MESSAGE: " + short(phase_msg[-1], t))
todo, mine, shown = [], [], set()
for m, age, pend in M.open_threads(log, t):
if m["type"] == "CALL":
topic = os.path.basename(os.path.dirname(next((r["path"] for r in m.get("refs", []) if isinstance(r, dict)), "x/x")))
voters = M.ballot_voters(M.load_ballot(topic)[0], ros)
if role in voters and not os.path.isfile(os.path.join(M.BOARD, "votes", topic, f"{role}.yaml")):
shown.add(m["id"])
todo.append(f"VOTE {topic}: write board/votes/{topic}/{role}.yaml (method/templates/vote.yaml), then "
f"post.py --from {role} --to CONVENER --type VOTE --re {m['id']} --ref board/votes/{topic}/{role}.yaml --text \"cast\" [{short(m, t)}]")
elif m["type"] == "VETO":
if M.addressed(m, role):
shown.add(m["id"])
todo.append("VETO (redesign or mitigate until SAFE lifts it): " + short(m, t))
elif M.pending_for(pend, m, role):
shown.add(m["id"])
todo.append(("OVERDUE " if age > M.STALL_MIN else "") + short(m, t) + f" -> answer: --type DECISION|ANS --re {m['id']}")
if m.get("from") == role and m["type"] in ("Q", "CR", "OBJECTION"):
mine.append(short(m, t))
print(f"TO DO ({len(todo)}):" + ("" if todo else " nothing addressed to you is open"))
for line in todo:
print(" " + line)
# everything else addressed to you (answers, decisions, INFO, FYI, tallies, nudges ...) since your last inbox
state = os.path.join(M.BOARD, f".inbox-{role}.json")
try:
seen = 0 if a.all else int(json.load(open(state, encoding="utf-8")).get("seen", 0))
except (OSError, ValueError, AttributeError):
seen = 0
new = [m for m in log if M.idn(m) > seen and m["id"] not in shown and m["type"] not in ("VOTE", "DONE")
and m.get("from") != role and M.addressed(m, role)]
cap = len(new) if a.all else 30
print(f"NEW FOR YOU ({len(new)}" + (", since your last inbox" if seen else "") + "):" + ("" if new else " nothing"))
if len(new) > cap:
print(f" ({len(new) - cap} older entries not shown: post.py inbox --role {role} --all)")
for m in new[-cap:]:
print(" " + short(m, t) + (f" (re {m['re']})" if m.get("re") else ""))
try:
json.dump({"seen": max([M.idn(m) for m in log] or [0]), "at": M.iso(t)}, open(state, "w", encoding="utf-8"))
except OSError:
pass
_, probs, _ = M.facts_check()
own = [p for p in probs if p[1] == role]
if own:
print(f"YOUR FACTS ({len(own)}): fix them in board/facts/{role}.yaml")
for sev, _, text in own[:20]:
print(f" {sev} {text}")
if mine:
print(f"WAITING ON OTHERS ({len(mine)}): keep working on a logged ASSUMPTION")
for line in mine:
print(" " + line)
def main():
if len(sys.argv) > 1 and sys.argv[1] in ("inbox", "show", "clock"):
ap = argparse.ArgumentParser(prog="post.py " + sys.argv[1])
if sys.argv[1] == "inbox":
ap.add_argument("--role", required=True)
ap.add_argument("--all", action="store_true", help="list everything addressed to you, not only what is new")
elif sys.argv[1] == "show":
ap.add_argument("id")
a = ap.parse_args(sys.argv[2:])
if sys.argv[1] == "clock":
print(M.clock_line())
else:
(cmd_inbox if sys.argv[1] == "inbox" else cmd_show)(a)
return
ap = argparse.ArgumentParser(description="append a message to board/log.jsonl and print the ask payloads")
ap.add_argument("--from", dest="sender", required=True)
ap.add_argument("--to", required=True)
ap.add_argument("--type", required=True)
ap.add_argument("--re")
ap.add_argument("--ref", action="append")
ap.add_argument("--text", required=True)
cmd_post(ap.parse_args())
if __name__ == "__main__":
main()
files/method/roles/ARCH.md (69 wierszy)
# Role ARCH - architecture and integration Mission: make the parts one product: the function structure, system alternatives assembled from every peer's fragments, the chosen architecture with its interfaces and budgets, the integrating concept document and the design report. You own `deliverables/02-concept/concept.md`, `physics-map.md`, `architecture.*`, `interfaces.md`, `deliverables/REPORT.md`, `board/concepts.md`, the concept ballot `board/votes/ARCH-concept/BALLOT.yaml` and, at the full tier, `deliverables/03-cad/assembly*` and `deliverables/03-cad/parts.csv`. Handbook sections 4, 5 and 15 (section 7 for the assembly) set the quality bar. ## Concept tier FRAME 1. Function structure in `board/ideas/ARCH.md`: the main functions as verb-noun pairs and the energy, material and signal flows; INFO to ALL so every fragment names the functions it serves. 2. Two fragments of your own at system level (where the subsystem boundaries lie, integrated versus modular, which function is done in hardware and which in software). 3. Budgets as facts `arch.budget.<name>` (mass, power, volume or cost) from the requirement thresholds or targets, label SOURCE with the requirement id; owners add `budget:` contributions. 4. `physics-map.md` from the convener's `board/SCAN.md` and the owners' physics tables; keep it current. 5. A `REPORT.md` skeleton with the handbook section 15 headings, marked as in progress, so the package always has a report. CONVERGE (start at once) 1. Within 7 minutes `board/concepts.md`: at least three system alternatives `A1..An` that differ in working principle or architecture, built as a morphological table (function -> the fragment each alternative uses, crediting its owner); for each: principle, block sketch, first-order numbers against the top-weighted requirements as fact ids, main risks and hazards. Missing numbers: one Q per owner with a 5-minute deadline, meanwhile a labelled ASSUMPTION. 2. The ballot from `method/templates/ballot.yaml` (kind concept, requirements all, RISK and SAFETY as in POLICY.md section 3, voters = every peer, closes = 8 minutes later), then CALL to ALL `--ref` the ballot. 3. Write your own vote before opening any other vote file. 4. On the TALLY: DECISION `--re` the CALL implementing the winner (the next one if SAFE vetoed it) and `board/decisions/D-ARCH-01.md` (winner, why each loser lost, sensitivity, contested ratings, SAFE's conditions); INFO to ALL with what each owner now writes for the chosen concept. WRITE (the busiest box: start concept.md and REPORT.md from your skeletons at once; you may draft parts of concept.md with subagents given self-contained briefs, but REPORT.md you write yourself) 1. `concept.md` in the order of handbook section 5. It integrates, it does not duplicate: summarise each owner's `sections/<ROLE>.md` and link it, quote numbers from the facts with their homes, take the alternatives from `board/concepts.md` and the comparison from the tally (knock-outs, weighted ratings, sensitivity, why each loser lost; details in `mesh/votes.md`), list decisions taken without a human with a pointer to `mesh/decision-log.md`. 2. The architecture block diagram as Mermaid in concept.md and `architecture.mmd`, plus a rendered SVG or PNG if a toolchain route exists (a short matplotlib script is enough). 3. `interfaces.md`: every interface between subsystems (mechanical, electrical, thermal, fluid, data, user) with both owners, the agreed numbers as fact ids and the `le`/`ge` constraints that check them. 4. Integrated key figures as facts with `uses:` of their contributions: `arch.dimensions` [L, W, H] mm, `arch.mass` kg, `arch.power` W. 5. `REPORT.md` per handbook section 15: quoted numbers only; status per requirement from REQ's compliance.csv; cost from COST's facts; top risks from the sections; decisions and blocked items from the log. Re-read it against `mesh/facts.md` before DONE. CLOSE: after the convener's export, check REPORT.md and concept.md against the facts once more; DONE final. ## Full tier - WRITE: `interfaces.md` v1 is the interface freeze; later changes need a CR to you and to the owners hit. - BUILD: an assembly script placing MECH's parts and the envelopes of purchased parts; `assembly.step` and `parts.csv` (`part,file,material,density_kg_m3,qty`); the interference check of `INPUT/TOOLCHAIN.md` on the assembly, clashes to MECH by CR; update `arch.dimensions` and `arch.mass` from CAD; hand MECH the assembly for the GLB and the renders. - CROSSCHECK: envelope against the CAD box, board outline against its bay, BOM against parts, power budget; CR the owners. - FINISH: the full REPORT.md with the file map; concept.md updated where detailed design changed it. ## Interfaces you negotiate Every domain owner (fragments, numbers, interfaces), REQ (requirements and statuses), COST (cost roll-up), SAFE (safety concept, vetoes).
files/method/roles/COST.md (39 wierszy)
# Role COST - sourcing, cost and price Mission: anchor the design in real parts and money: the benchmark, price anchors, cost per alternative, the unit-cost roll-up and the retail price logic; the full BOM at the full tier. You own `deliverables/02-concept/benchmark.md`, `deliverables/02-concept/sections/COST.md`, your `cost_*` calculation scripts and `deliverables/09-bom-cost/` (concept tier: optional cost.json; full tier: bom.csv, cost.json, cost.md in the RUN CONTRACT formats). Handbook sections 3 and 12 set the quality bar. ## Concept tier FRAME 1. `benchmark.md`: at least five comparable products (commercial and open source; adjacent categories count) with the specifications that matter for our requirements, price with date and URL, and weak points from reviews, teardowns and return reasons; from them the realistic targets and our differentiator. Summary in `board/ideas/COST.md`, INFO to ALL. 2. Price anchors as facts: target retail price range, target unit cost at 1, 100 and 1000, typical prices of the key part classes from the offline catalogue and distributor pages (SOURCE with the date read). CONVERGE: a rough cost per alternative as facts (`cost.alt_<id>_q1000` with the method in the source), then vote (your criteria and RISK: supply risk, single sources, lead times). WRITE: `sections/COST.md`: a rough BOM of the major items with real candidates; the unit cost at 1, 100 and 1000 with every line of handbook section 12 (parts, board fabrication and assembly, mechanics by process, labour, packaging, certification and tooling amortised over a stated quantity, freight, contingency); the retail price logic (margin, platform and payment fees, VAT, shipping, returns) and a plain statement whether the numbers close. Key-figure facts `cost.unit_cost_q1`, `cost.unit_cost_q100`, `cost.unit_cost_q1000`, `cost.retail_price` (unit = currency code). Optionally `09-bom-cost/cost.json` in the RUN CONTRACT format. ## Full tier - BUILD: `bom.csv` with exactly the RUN CONTRACT header and real purchasable parts (catalogue first, distributor pages read-only; the price date in `notes`, otherwise ASSUMPTION), every product variant the requirements ask for, second sources for critical parts; `cost.json` and `cost.md` consistent with bom.csv (the checker compares the sums). - CROSSCHECK: every CAD part and every schematic part has a BOM line (the convener's gate lists gaps); item numbers agreed with MECH's assembly drawing. ## Interfaces you negotiate MECH and ELEC (parts, processes, quantities), REQ (price and cost requirements), ARCH (cost budget).
files/method/roles/DOMAIN.md (29 wierszy)
# Role DOMAIN - a domain peer added by the physics scan The convener added you because its physics scan rated a domain CRITICAL that neither MECH nor ELEC covers in depth (for example optics, acoustics, chemistry, radio or motion control). Your role code and your domain are in your spawn task; below, `<CODE>` means that code and `<code>` the same in lower case. Mission: be the authority on your domain: its physics, its alternatives, its numbers and its critical parts. You own `deliverables/02-concept/sections/<CODE>.md`, your `<code>_*` calculation scripts and your board files. Parts of other deliverables that your domain needs (a CAD part, a circuit block, a firmware module, a BOM line) are produced by their owners from your specification: send CRs with exact numbers and record the interface as facts with `le`/`ge` constraints. ## Concept tier FRAME: the physics table of your domain in `board/ideas/<CODE>.md`, deeper than the handbook minimum (governing relations, dominant parameters, order-of-magnitude numbers, failure modes from first principles and from user reports); two fragments with different working principles, with first-order numbers from short scripts; the parts that decide feasibility as real candidates (part number, price with date). CONVERGE: numbers for ARCH within 5 minutes; vote (your criteria and RISK). WRITE: `sections/<CODE>.md`: the domain design of the chosen concept, estimates with margins, parts, risks, what detailed design must do first; final facts. ## Full tier - BUILD: the calculations and simulations of your domain in `deliverables/05-calculations/<code>_*.py` (known-answer case first) and complete specifications for the owners who implement your parts. - CROSSCHECK: their implementation against your numbers; CRs where it deviates.
files/method/roles/ELEC.md (43 wierszy)
# Role ELEC - electronics, power and firmware Mission: power, sensing, control and firmware: the electrical architecture and power budget at the concept tier; the KiCad project, circuit checks and the PlatformIO firmware at the full tier. You own `deliverables/02-concept/sections/ELEC.md`, your `elec_*` calculation scripts and, at the full tier, `deliverables/06-electronics/` and `deliverables/07-firmware/`. Handbook sections 4, 6, 9 and 10 set the quality bar. ## Concept tier FRAME 1. Physics table in `board/ideas/ELEC.md`: electrical power, electronics and control, firmware, radio and EMC. 2. Two fragments with different electrical or control architectures (supply topology, sensing principle, control method), each with a power budget over the full supply range and every operating mode (facts with `budget: power`) and the key parts as real candidates: the offline catalogue of `INPUT/TOOLCHAIN.md` first (part number, stock, price with date). 3. For every hazard SAFE raises in your domain: the hardware protection that works without firmware. CONVERGE: numbers for ARCH within 5 minutes; vote (your criteria and RISK). WRITE: `sections/ELEC.md`: electronics block diagram, power budget table, protection concept (fuse or current limit, reverse polarity, transients and ESD, inrush, thermal cut-offs in hardware, safe state on power loss), firmware architecture (state machine with fault states, safe default, watchdog), key parts table, radio and EMC approach (pre-certified modules), risks. Final facts, including the board size with a `le:` constraint against MECH's bay. ## Full tier - BUILD: KiCad 9 project (schematic, and a routed board where custom electronics is needed) with real parts and matching footprints; ERC and DRC as JSON reports, every error fixed or its waiver justified; fabrication outputs; ngspice where circuit behaviour decides something (known-answer test first); `deliverables/05-calculations/elec_*.py` (currents, dissipation and temperature rise, trace widths, creepage and clearance of mains parts with the standard as SOURCE); the PlatformIO project at the root of `07-firmware/` building offline for the schematic's MCU, with native host tests of the logic and the interlocks (a test fails first, then passes) and the pin map checked against the schematic. - CROSSCHECK: board outline against MECH's bay, connectors against enclosure openings, BOM lines with COST, safety functions with SAFE. - FINISH: rebuild the firmware and rerun ERC and DRC on the final files. ## Interfaces you negotiate MECH (board space, heat, actuator and heater power), SAFE (protections, insulation, standards), COST (parts, second sources), ARCH (power budget, user interface).
files/method/roles/MECH.md (47 wierszy)
# Role MECH - mechanics, heat and fluids Mission: own the physical product (structure, layout, heat, fluids, mechanisms, materials), from first-order physics at the concept tier to CAD, drawings, calculations and renders at the full tier. You own `deliverables/02-concept/sections/MECH.md`, your `mech_*` calculation scripts and, at the full tier, `deliverables/03-cad/` (except ARCH's assembly files), `04-drawings/` and `11-renders/`. Handbook sections 4, 6, 7, 8 and 14 set the quality bar. ## Concept tier FRAME 1. Physics table for your domains in `board/ideas/MECH.md` (`method/templates/idea.md`): structure, heat, fluids, mechanisms, materials, and also acoustics, human factors and environment unless another peer covers them: rating, governing relations, dominant parameters, order-of-magnitude numbers, how it typically fails. 2. Two fragments that differ in working principle or layout, each with first-order numbers: a hand estimate first, then a short script `deliverables/02-concept/calc/mech_<name>.py` (run with `ddpy`, writes its results.json); publish the numbers as facts. 3. What you need from others as numbers, by CR or Q (heating or drive power from ELEC, the space the electronics needs, prices from COST). CONVERGE: answer ARCH's questions with numbers within 5 minutes; vote (your criteria and RISK). WRITE: `sections/MECH.md` for the chosen concept: layout with key dimensions; each governing estimate as value, requirement, margin, status; materials and processes at 1, 100 and 1000 units; the critical mechanical and purchased parts with real candidates (manufacturer, part number, price with date); what detailed design must do first; risks. Final facts: mass and size contributions with `budget:`, interface constraints with ELEC and the domain peers. ## Full tier - BUILD: parametric code-CAD, one script per part with its parameters at the top taken from the facts, one STEP per part; check every part (valid solid, mass from density against your mass facts, bounding box against the size requirements, walls and drafts for the process, standard fasteners); a drawing of every manufactured part and the assembly drawing with balloons matching COST's bom.csv item numbers (agree them by CR); calculations `deliverables/05-calculations/mech_*.py` for what the physics map rates CRITICAL (known-answer case first; a simulation only where it changes a decision, with a mesh or step study). - CROSSCHECK: ELEC's board against your bay, COST's BOM against your parts, SAFE's hazards against your design measures; CRs to the owners. - FINISH: the GLB of ARCH's assembly (the STEP to GLB converter of `INPUT/TOOLCHAIN.md`) and renders in Blender background mode (product view, exploded view, user-interface view), checked by measurement (file, resolution, framing, part names), not by eye. ## Interfaces you negotiate ELEC (board size and position, heat from electronics, power of actuators and heaters), ARCH (envelope, assembly), SAFE (guards, temperatures, materials near heat and live parts), COST (parts and processes).
files/method/roles/REQ.md (59 wierszy)
# Role REQ - requirements and verification
Mission: turn the input into requirements that an engineer can design and verify from, keep the compliance
table honest, and at the full tier plan the tests that would prove every requirement.
You own `deliverables/01-requirements/` (requirements.md, compliance.csv, optional requirements.yaml),
`board/reqmap.yaml` and, at the full tier, `deliverables/10-prototype-test-plan/`. Handbook sections 1, 2,
13 and 16 set the quality bar; the RUN CONTRACT fixes the compliance.csv format.
## Concept tier
FRAME
1. Within 10 minutes write `board/reqmap.yaml`: every requirement id of `INPUT/requirements.md` mapped to
the role(s) whose design produces its evidence, as `ID: [ROLE]` lines (usually one role: size, mass,
heat, fluids and mechanics to MECH; power, electronics, control, firmware and radio to ELEC; safety and
certification to SAFE; cost and price to COST; user experience and system-level items to ARCH; items only
you can verify to REQ; a domain peer's items to its code; a merged role's items to its code or its
holder). Check with `python3 method/gates/board_check.py reqs` (no ERROR, no UNMAPPED), then INFO to ALL:
"reqmap ready".
2. At once write `compliance.csv` with one row per id (`not_addressed`, empty evidence) so the package is
never invalid; refine it as evidence appears.
3. Draft `requirements.md`: (1) the given requirements with your interpretation (the exact condition under
which each is verified) and its design consequence; (2) derived requirements `DR-<AREA>-<nn>` walking
every area of handbook section 2 (say why an area does not apply); (3) conflicts and decisions, and the
assumptions register. Owners send you CRs for derived requirements of their domains: you decide (DECISION
when accepted, ANS with the reason when not) and cite the message id in the requirement.
4. Publish the numeric verification conditions you fix (ambient range, supply range, duty cycle, test
durations) as facts `req.*` in `board/facts/REQ.yaml`; owners design to them and list them under `uses:`.
5. Tell the owners concerned (one INFO each) which requirement pairs pull against each other, so their
fragments address the conflict.
CONVERGE: vote. You are the threshold screen: for every option compare the predicted values of the
top-weighted requirements (`board_check.py reqs` lists the facts that cite each id) with the thresholds and
put real, evidenced misses in `ko`, the requirement id first (your knock-outs always count; another peer's
only for requirements mapped to it). Rate your assigned criteria and RISK.
WRITE
1. Finish `requirements.md` for the chosen concept.
2. Fill `compliance.csv`, one row per input id: `met` (target or threshold met with evidence), `partly`
(threshold met but target missed, or evidence incomplete), `not_met` (predicted miss, with the reason),
`not_addressed` (no evidence); `value` and `unit` in the requirement's own metric from the owner's fact;
`evidence` = existing paths inside `deliverables/` (sections, calc results, concept.md), never `board/`;
`note` = how the value was obtained ("predicted, CALCULATION"). A value without a fact: Q to its owner.
3. `python3 method/gates/dd_check.py --tier concept` and fix every finding in your files.
CLOSE: re-check every row against the final facts; DONE `--ref final`.
## Full tier
- BUILD: keep compliance.csv current as evidence arrives (CAD masses, calculations, ERC/DRC, firmware tests).
Write `10-prototype-test-plan/`: build plan of prototype 1 (with MECH and ELEC), bring-up with a
current-limited or isolated supply and the measurements before full power, then one test per requirement
and per safety requirement (method, equipment, conditions, pass criterion, requirement id), pre-compliance
EMC, the lab tests SAFE lists, durability tests where lifetime is a requirement.
- CROSSCHECK: trace every id to evidence; CR the owners whose evidence is missing or contradicts the facts.
- FINISH: final statuses; the test plan covers every id.
## Interfaces you negotiate
Every owner (numbers for compliance by Q; their derived requirements by CR to you), SAFE (safety
requirements and lab tests), COST (price and cost requirements).
files/method/roles/SAFE.md (43 wierszy)
# Role SAFE - safety and compliance (holds the veto) Mission: make the product safe by design and legally sellable: hazards, safety requirements, protections and the regulatory route. You hold the only veto in the mesh. You own `deliverables/02-concept/sections/SAFE.md`, your `safe_*` calculation scripts and, at the full tier, `deliverables/08-safety-certification/`. Handbook sections 11 and 16 set the quality bar. ## Concept tier FRAME 1. Hazard list in the ISO 12100 manner for this product and its foreseeable use and misuse (hazard, hazardous situation, harm, severity x probability) in `board/ideas/SAFE.md`; as fragments appear, tell each owner which hazards its fragment creates (INFO or CR). 2. Regulatory map with SOURCE: the EU directives and the harmonised product standards that apply to this product (found by research, with the clauses that drive the design), the US rules, the likely conformity route. Derived safety requirements `DR-SAF-<nn>` go to REQ as CRs; design limits (surface temperatures, creepage, stability) become your facts with their sources. CONVERGE: rate SAFETY for every option and your assigned criteria. Put an option in `veto` only for a hazard of serious or worse severity that no proportionate measure brings to an acceptable residual risk, and name the condition that would lift it. WRITE: `sections/SAFE.md`: the top hazards with their measures in the order inherently safe design, protective devices, information for use; which protections act independently of firmware; residual risks; the regulatory route with SOURCE, the conformity route, the tests an accredited lab would run, a rough cost and duration. You identify and plan; you never certify. ## Veto discipline VETO a decision or option with: the hazard, its severity, why the current measures are insufficient, the lift condition. LIFT `--re` the VETO as soon as the condition is met. Preferences that are not safety are OBJECTIONs, never vetoes. ## Full tier - BUILD: `08-safety-certification/`: the hazard analysis with residual risks, safety requirements traced to design measures and tests, the directives and standards with the clauses that drive the design, the conformity route, a pre-compliance plan, the lab test list with rough cost and duration; `deliverables/05-calculations/safe_*.py` where a number decides safety. - CROSSCHECK: every hazard's measure exists in the CAD, the schematic or the firmware (CR the owner if not); REQ's test plan contains every safety test. ## Interfaces you negotiate Every owner (the hazards their design creates and the measures), REQ (safety requirements and tests).
files/method/templates/ROSTER.md (22 wierszy)
# ROSTER - who writes what (single writer per path) Written by the convener (`board_check.py init`, then `board_check.py roster --add/--merge/--set/--move`). Peers find each other's session ids here. A path belongs to the role whose pattern matches it most specifically (the longest matching pattern wins). Besides the paths below, every peer owns its board files and its calculation scripts: `board/facts/<ROLE>.yaml`, `board/ideas/<ROLE>.md`, `board/decisions/D-<ROLE>-*`, `board/votes/*/<ROLE>.yaml`, `board/votes/<ROLE>-*/BALLOT.yaml`, `board/out/<ROLE>-*`, `deliverables/02-concept/sections/<ROLE>.md`, `deliverables/02-concept/calc/<role>_*`, `deliverables/05-calculations/<role>_*` (`<role>` = the code in lower case). `board/log.jsonl` is written only through `method/gates/post.py`. Status: pending, live, degraded (a stand-in subagent runs the role), dead, done. | role | session | status | brief | owns | |---|---|---|---|---| | CONVENER | - | live | MAIN.md | board/ROSTER.md, board/CLOCK.json, board/STATUS.md, board/SCAN.md, board/RUN-CONTRACT.md, board/votes/*/TALLY.md, deliverables/manifest.json, deliverables/02-concept/mesh/* | | REQ | PENDING | pending | method/roles/REQ.md | deliverables/01-requirements/*, deliverables/10-prototype-test-plan/*, board/reqmap.yaml | | ARCH | PENDING | pending | method/roles/ARCH.md | deliverables/02-concept/concept.md, deliverables/02-concept/physics-map.md, deliverables/02-concept/architecture*, deliverables/02-concept/interfaces.md, deliverables/REPORT.md, deliverables/03-cad/assembly*, deliverables/03-cad/parts.csv, board/concepts.md | | MECH | PENDING | pending | method/roles/MECH.md | deliverables/03-cad/*, deliverables/04-drawings/*, deliverables/11-renders/* | | ELEC | PENDING | pending | method/roles/ELEC.md | deliverables/06-electronics/*, deliverables/07-firmware/* | | SAFE | PENDING | pending | method/roles/SAFE.md | deliverables/08-safety-certification/* | | COST | PENDING | pending | method/roles/COST.md | deliverables/02-concept/benchmark.md, deliverables/09-bom-cost/* |
files/method/templates/ballot.yaml (18 wierszy)
# board/votes/<CALLER>-<topic>/BALLOT.yaml - written by the caller BEFORE the CALL and frozen by it
# (the tally refuses silently changed ballots). The folder name is the topic: ARCH-concept, MECH-joint-type ...
topic: ARCH-concept
kind: concept # concept (system alternatives) | dispute (an owner's decision is contested)
question: "Which system concept do we take into design?"
options: # id: one line; the full description lives in board/concepts.md or a file you own
A1: "..."
A2: "..."
A3: "..."
requirements: all # concept: every requirement id is a criterion (weight = its priority in the
# input, raters = board/reqmap.yaml); dispute: none
extra_criteria: # concept: these two (POLICY.md section 3); dispute: replace by
RISK: {share: 0.15, roles: ALL} # SAFETY {weight: 5}, COMPLIANCE {weight: 4}, COST {weight: 3},
SAFETY: {share: 0.10, roles: [SAFE]} # SCHEDULE {weight: 2}, SIMPLICITY {weight: 1}, all with roles: ALL
voters: ALL # concept: ALL (every peer in board/ROSTER.md); dispute: the affected owners + SAFE,
# e.g. [MECH, ELEC, SAFE]
closes: "HH:MM" # quoted clock time; the convener tallies when all voted or at this time
files/method/templates/decision.md (8 wierszy)
# D-<ROLE>-<nn> - <short title> - Owner: <ROLE> (owner of: <path or fact ids>) - logged as <M-id> (DECISION), settles <M-id> (CR / Q / TALLY / OBJECTION) - Decision: <what is now true, with numbers and fact ids> - Alternatives rejected and why: <each with the evidence or the POLICY.md rule that decided> - Evidence: <calc results, facts, requirement ids, SOURCE pages> - Consequences: <which owners must update which files or facts; the CRs or INFO messages you sent them>
files/method/templates/facts.yaml (31 wierszy)
# board/facts/<ROLE>.yaml - every design number you own, one entry each. Only you write this file.
# id: <your role code in lower case>.<name> (mech.enclosure_mass, elec.idle_power, cost.unit_cost_q1000)
# value: a number or a list of numbers ([L, W, H]); unit: mm, kg, W, N, EUR ... ("1" if dimensionless)
# label: SOURCE | CALCULATION | SIMULATION | MEASUREMENT | ASSUMPTION
# source: the number's home - script or results file (CALCULATION), URL + date read (SOURCE),
# result file (SIMULATION), or for an ASSUMPTION the reason and how sensitive the design is to it.
# Optional: req (requirement ids it is evidence for), budget (adds to arch.budget.<name>),
# le / ge (must be <= / >= that fact: an interface agreement), note (one line).
# Key-figure ids (single home of the manifest figures): arch.dimensions [L, W, H] mm, arch.mass kg,
# arch.power W (ARCH); cost.unit_cost_q1 / _q100 / _q1000 and cost.retail_price, unit = currency (COST).
owner: ROLE
facts:
- id: role.example_power
value: 12.5
unit: W
label: CALCULATION
source: deliverables/02-concept/calc/role_power.py
req: [ABC-PWR-01]
budget: power
note: worst case at the highest supply voltage
- id: role.example_width
value: 80
unit: mm
label: ASSUMPTION
source: "typical for this class (benchmark median); +/-15 mm moves the enclosure size only"
le: other.bay_width
uses: # numbers of other owners you rely on, with the value you used
- id: other.bay_width
value: 85
unit: mm
files/method/templates/idea.md (16 wierszy)
# Ideas - <ROLE> ## Physics of my domains (handbook section 4) | domain | rating | governing relations | dominant parameters | first-order numbers (fact ids) | typical failures | |---|---|---|---|---|---| ## F-<ROLE>-1 - <title: a working principle or layout, not a cosmetic variant> - Working principle: - Functions served (ARCH's function list): - First-order numbers: <fact ids and the calc script behind them> - Needs from other owners (interfaces): <power, space, heat, data ... as numbers> - Risks and hazards: - Combines well with: <fragment ids of other owners> ## F-<ROLE>-2 - <a genuinely different alternative>
files/method/templates/vote.yaml (18 wierszy)
# board/votes/<topic>/<ROLE>.yaml - your blind vote. Write it BEFORE opening any other file of the topic
# folder except BALLOT.yaml, then log: post.py --from <ROLE> --to CONVENER --type VOTE --re <CALL id>
# --ref board/votes/<topic>/<ROLE>.yaml --text "cast". Do not discuss ratings until the TALLY.
# Scale per criterion: 0 = misses the threshold, 5 = meets the threshold, 8 = meets the target,
# 10 = beats the target with margin. RISK and SAFETY: 10 = lowest risk / hazard, 0 = unacceptable.
# Rate every criterion assigned to your role (requirement criteria: board_check.py reqs; extra ones: the
# ballot's roles); others only with evidence. Every section is a mapping (key: value), never a list.
voter: ROLE
ratings: # criterion -> {option: 0..10}
ABC-PRF-01: {A1: 7, A2: 5, A3: 9}
RISK: {A1: 6, A2: 8, A3: 4}
reasons: # criterion -> one line with the evidence (fact ids, calc results)
ABC-PRF-01: "A3 meets the target with 20 % margin (mech.stiffness_margin); A2 only the threshold"
RISK: "A3 needs a part with a single source (cost.motor_sources)"
ko: {} # option -> "REQ-ID: predicted value vs threshold (fact id)" - a predicted miss; it counts
# from REQ or from a rater of that requirement in board/reqmap.yaml
veto: {} # SAFE only: option -> "hazard id, severity, why no mitigation is acceptable"
skills/dd-hw-handbook/SKILL.md (266 wierszy)
---
name: dd-hw-handbook
description: Product-agnostic quality bars and methods for every deliverable of a dd hardware design run (requirements, concept, calculations, CAD, drawings, electronics, firmware, safety and certification, BOM and cost, prototype and test plan, renders, report). Read it before producing any deliverable. It says WHAT GOOD LOOKS LIKE; your main prompt says who does the work and in which order.
---
# dd hardware design handbook
Your main prompt defines the team, the roles, the communication and the order of work. This handbook defines
the quality bar of each deliverable and the methods that reach it. Where your main prompt is more specific,
follow your main prompt. Where the RUN CONTRACT fixes a file name or a format, the RUN CONTRACT wins.
Exact tool commands, versions and paths are in `INPUT/TOOLCHAIN.md` (read the working examples it points to
before writing your own scripts); this handbook names tools only by purpose.
Contents: 1 inputs · 2 requirements · 3 research · 4 physics map · 5 concept · 6 calculations · 7 CAD ·
8 drawings · 9 electronics · 10 firmware · 11 safety and certification · 12 sourcing, BOM and cost ·
13 prototype and test plan · 14 renders · 15 report and manifest · 16 claims · 17 time and context.
---
## 1. Reading the inputs
- `INPUT/brief.md` — one sentence plus customer context. Extract: who buys it, where and how it is used, how
often and how long, the price expectation, the one or two things that must be better than what exists.
- `INPUT/requirements.md` — the acceptance criteria. Each item has an id, a metric, a target and usually a
threshold (worst acceptable), a verification method and a weight. **Every id must be addressed and keep its
exact spelling** wherever you mention it. Weights tell you what matters most when requirements conflict.
- `INPUT/TOOLCHAIN.md` — what is installed and how to call it headless. Use only what it lists. A tool that
misbehaves gets three honest attempts with new information each time, then a labelled fallback (for example
an analytical model instead of a simulation) and a note in `manifest.json → blocked[]` if it mattered.
## 2. Requirements — `deliverables/01-requirements/`
Quality bar: an engineer who has never seen the brief can design and verify the product from this folder
alone, and every number in it has an origin.
Write `requirements.md` (and, if you like, a machine-readable `requirements.yaml`) with three parts:
1. **Given requirements**, one entry per id from `INPUT/requirements.md`: the id, the original statement, your
interpretation (the exact condition under which it will be verified: temperature, voltage, distance,
duty cycle, test method), and the design consequence (what it forces in the design).
2. **Derived requirements**, ids `DR-<AREA>-<nn>`, each with: parent id(s) or "brief"/"regulation"/
"manufacturing", statement, metric, target, threshold, verification method (Analysis, Inspection,
Demonstration, Test), rationale with origin label, priority. Walk at least these areas and write down
either the requirement or why the area does not apply: function and performance under all operating
conditions; user interface and ergonomics; power source and consumption (including standby); mechanical
interfaces and mounting; environment (temperature, humidity, dust and water, altitude, transport and
storage); safety (electric shock, heat and burns, fire, mechanical hazards, chemical, optical and other
radiation, noise, misuse that can be foreseen); EMC and, if any radio, radio; materials and restricted
substances; lifetime, duty cycle and reliability; maintenance, cleaning, consumables and repair;
manufacturing process at 1, 100 and 1000 units; assembly and test in production; packaging, shipping and
drop; labelling, manuals and markings; unit cost and retail price; prototype schedule.
3. **Conflicts and decisions**: requirement pairs that pull against each other, how you resolved each (use the
weights; safety first), and the assumptions register (each ASSUMPTION with reason and sensitivity).
Write `compliance.csv` exactly as the RUN CONTRACT specifies, one row per id of `INPUT/requirements.md`:
- `status`: `met`, `partly`, `not_met` or `not_addressed`. At the concept tier the status is a prediction
from your estimates: say so in `note` (for example "predicted, CALCULATION"). Be honest: a predicted miss
is `not_met` with the reason, not a hidden problem.
- `value` and `unit`: the predicted or designed value in the requirement's own metric and unit.
- `evidence`: path(s) inside `deliverables/` that show it (a calculation result, a section of the concept).
## 3. Research — benchmark, users, prior art
- Find at least five comparable products (commercial and open-source; adjacent categories count). For each:
the specifications that matter for our requirements, price with date and URL, and weak points from
reviews, teardowns, forums or return reasons. Put the table in `02-concept/benchmark.md`.
- From the benchmark derive: realistic target values, the differentiator ("our own, better version" — what we
do better and why it is feasible), and the failure modes users actually complain about. Feed both back into
the derived requirements.
- Note patents only as prior art you noticed; never give legal opinions. If an idea of yours looks genuinely
new, flag it as `novelty` in `REPORT.md` (one line) — do not elaborate on it in public-facing text.
- Read pages, do not guess them. If a site blocks automated reading, use a different source.
## 4. Physics map — which disciplines decide this product
Rate each domain CRITICAL, RELEVANT or MINOR for this product: structure and strength; heat; fluids and
aerosols; electrical power; electronics and control; firmware; mechanisms and motion; materials and
chemistry; optics and light; acoustics; human factors; environment. For each CRITICAL domain write: the
governing relations, the few parameters that dominate the result, the order-of-magnitude numbers, and the
ways it typically fails (from first principles and from what users report). The map decides which analyses
you run and how deep; it lives in `02-concept/physics-map.md`.
## 5. Concept — `deliverables/02-concept/`
Quality bar: the decision is visibly earned. A reader sees real alternatives, numbers for each, a fair
comparison, and a chosen architecture concrete enough to start detailed design tomorrow.
Contents (file names are suggestions; keep them inside `02-concept/`):
- `concept.md` — the main document, in this order:
1. The product in one paragraph: what it is, for whom, what it does better.
2. Functions and flows: the main functions as verb–noun pairs and the energy, material and signal flows.
3. Alternatives: at least three that differ in **working principle or architecture**, not in cosmetics.
For each: how it works, a simple diagram, first-order numbers against the top-weighted requirements
(performance, power, size, mass, unit cost), the main risks and hazards.
4. Evaluation: knock-out by thresholds first; then a weighted comparison whose weights come from the
requirement weights; a sensitivity check (vary each weight ±20 %, and the most uncertain estimates
within their range) showing whether the winner holds. Record why each loser lost.
5. The chosen concept: architecture block diagram (Mermaid in the markdown and a rendered PNG or SVG if
the toolchain has a renderer), the subsystem list with make-or-buy, the critical components with real
candidate parts (manufacturer, part number, supplier link, price with date), the interfaces between
subsystems (mechanical, electrical, thermal, data).
6. Key estimates: every number the decision rests on, each with its origin; the scripts that compute them.
7. Safety concept: the top hazards, the design measures (inherently safe design first, then protective
devices, then information for the user), and which protections act **independently of firmware**.
8. Compliance route: which EU directives and harmonised standards, and which FCC rules, apply and why
(with SOURCE), the likely conformity route, rough cost and time.
9. Cost: a rough BOM of the major items, the unit cost at 1, 100 and 1000 units with the method, and the
retail price logic (section 12). Optionally `09-bom-cost/cost.json` already at this tier.
10. Risks: the top risks with likelihood, impact, mitigation and the analysis or test that would retire
each one.
11. Decisions taken without a human (taste, priorities, trade-offs) with the rejected alternatives.
12. Open issues and what detailed design must do first.
- `calc/` — the scripts behind the key estimates (same rules as section 6), with their results files.
- `benchmark.md`, `physics-map.md`, and anything your method produces (scores, votes, issue logs, models).
Concept anti-patterns: two strawmen next to the favourite; alternatives that differ only in colour or
material; scores without numbers behind them; a decision matrix whose weights were tuned until the favourite
won; a concept that silently misses a top-weighted requirement; prose without a single calculated number.
## 6. Calculations — `05-calculations/` (full tier) and `02-concept/calc/` (concept tier)
- One Python script per question, run with the toolchain's Python (`ddpy`, see `INPUT/TOOLCHAIN.md`; the
system `python3` has no scientific libraries). Physical quantities carry units (the `pint` library) and every
input is labelled at the top of the script with its origin.
- Scripts run with `ddpy <script>` from their own folder, offline, in under five minutes, and write
`<stem>.results.json` (`{"<name>": {"value": x, "unit": "..."}}`). The RUN CONTRACT requires this for
`05-calculations/`; use the same convention in `02-concept/calc/`.
- Order of work: hand estimate → script → (only if it changes a decision) simulation. For any simulation:
first a known-answer case, then the real case on at least three mesh or step sizes with the trend shown,
sanity checks (balances of mass, energy or force), and a sensitivity run on the least certain inputs.
- Static strength is not drop, impact or fatigue; steady thermal is not transient; say which remain open.
- Margins: state each result as value, requirement, margin, and status.
## 7. CAD — `03-cad/` (full tier)
- Parametric code-CAD (CadQuery or build123d, whichever `INPUT/TOOLCHAIN.md` lists as working), one script
per designed part, parameters at the top, an assembly script placing parts with names. Re-running the
scripts reproduces every STEP. FreeCAD files are acceptable if that is your route.
- Deliver `assembly.step` and one STEP per designed part, plus the source. Purchased parts appear as
simplified bodies with their real envelope dimensions from the datasheet (label them).
- Check every part: valid solid, volume and mass from density, bounding box against size requirements,
pairwise interference zero (or an intended fit), clearances for assembly tools and fingers, wall thickness
and draft suited to the chosen process, standard fasteners.
- Choose a manufacturing process per part for the prototype (3D printing, CNC, laser-cut sheet) and for
1000 units (injection moulding, die cutting, sheet-metal bending), and note what changes between them.
- Export a GLB of the assembly for the renders and the website (see `INPUT/TOOLCHAIN.md` for the converter).
## 8. Drawings — `04-drawings/` (full tier)
- A drawing for every manufactured part: views, overall and functional dimensions, tolerances (a general
tolerance class plus explicit tolerances on functional features), material, finish, quantity, part number
and revision matching the BOM. Flat patterns for sheet metal.
- An assembly drawing with item balloons that match `bom.csv` item numbers. PDF or SVG; DXF optional.
- Open each output after you write it (render a PNG preview and look at it, or parse it) — a drawing file
that does not open does not exist.
## 9. Electronics — `06-electronics/` (full tier)
- Start from a block diagram and a power budget over the full supply range and all operating modes.
- KiCad 9 project with schematic and, where custom electronics is needed, a routed board. Real parts with
matching symbols and footprints; every part has a manufacturer part number that exists at a distributor or in
the offline LCSC/JLC catalogue described in `INPUT/TOOLCHAIN.md`.
- Protection where it applies: input fuse or current limit, reverse polarity, transients and ESD on external
connectors, inrush, thermal shutdown of hot parts in hardware, safe state on power loss.
- Calculations: currents, trace widths, dissipation and temperature rise of the hot parts, derating, and for
any mains part the creepage and clearance with the standard you took the values from.
- Run ERC and DRC from the command line; deliver the reports; fix every error or justify each waiver.
Fabrication outputs: Gerbers, drill, pick-and-place and the board BOM. Passing DRC is not a hardware test.
- Simulate uncertain circuit behaviour with ngspice where it changes a decision (start-up, protection trip,
filter response) after one known-answer test of the simulator.
## 10. Firmware — `07-firmware/` (full tier)
- PlatformIO project at the folder root that builds offline for the real target MCU of the schematic; use
exactly the platform versions `INPUT/TOOLCHAIN.md` pins (anything else downloads packages and blocks
other runs).
- A state machine with explicit fault states and a safe default; a watchdog; inputs debounced; every
safety-relevant reaction has a hardware backstop and the firmware is never the only protection.
- Host tests (`native` environment) for the logic and the interlocks: write a test that fails first, then
passes. Build with warnings as errors where the framework allows. Never flash or power anything.
- Pin map agreed with the schematic; configuration constants generated from or checked against the design
parameters.
## 11. Safety and certification — `08-safety-certification/` (both tiers: summary in 02; full tier: complete)
- Hazard analysis in the ISO 12100 manner: hazard → hazardous situation → harm → risk estimate (severity ×
probability) → reduction in the order inherently safe design, protective measures, information for use →
residual risk. Derive safety requirements with ids and trace them to the design.
- Regulatory map with SOURCE for each item. For the EU think through at least: the Low Voltage Directive
(2014/35/EU) when any voltage is in its range, the EMC Directive (2014/30/EU), the Radio Equipment Directive
(2014/53/EU) when there is any radio, RoHS (2011/65/EU), WEEE (2012/19/EU), the batteries rules when there
is a battery, the General Product Safety Regulation (EU) 2023/988, REACH, ecodesign and standby rules.
For the US: FCC Part 15 (Subpart B for unintentional radiators, Subpart C for intentional ones) and the
voluntary safety listings retailers expect. Then find the harmonised product-family standards that
apply to THIS product by research, and quote which clauses drive the design.
- Prefer architectures that make conformity easy: low voltage inside the product through a certified external
supply when the power level allows it, pre-certified radio modules, protections that do not depend on
firmware, materials with suitable flammability ratings near heat and live parts.
- State the conformity route (self-declaration, notified body or not), the tests an accredited lab would run,
a pre-compliance plan, and a rough cost and duration. You identify; you do not certify.
## 12. Sourcing, BOM and cost — `09-bom-cost/`
- Real, purchasable parts only. For each line: manufacturer, part number, supplier, supplier part number, URL,
unit price at 1, 100 and 1000 (price breaks as shown by the supplier, currency stated), the date you read it
in `notes`. A price you could not read is an ASSUMPTION with a range, never a guess dressed up as SOURCE.
Prefer parts in stock at two or more suppliers, active (not end-of-life), and note a second source for
critical parts. Use the offline catalogue from `INPUT/TOOLCHAIN.md` first, then distributor pages
read-only.
- `bom.csv` with exactly the header the RUN CONTRACT gives; `cost.json` and `cost.md` as it specifies.
- The unit cost at each quantity is more than the BOM: PCB fabrication and assembly, mechanical parts (at 1
and 100 units printed or machined; at 1000 units moulded with the tooling cost amortised and stated
separately), assembly and test labour, packaging, certification and other one-off costs amortised over
the quantity (state the assumption), freight to the fulfilment point, and a contingency. Show each line.
- Retail price logic: start from benchmark prices and a target gross margin; check that the price leaves room
for crowdfunding platform and payment fees, shipping to backers, VAT where it applies, returns and warranty.
An underpriced product that cannot be delivered is the classic crowdfunding failure; say plainly if the
numbers do not close.
## 13. Prototype and test plan — `10-prototype-test-plan/` (full tier)
- Build plan for prototype 1: how each part is made, the order of assembly, tools, and the checks after each
step.
- Bring-up for a person later: first power-up with a current-limited or isolated supply, what to measure
before applying full power, safety precautions.
- One test per requirement (and per safety requirement): method, equipment, conditions, pass criterion, the
requirement id it closes. Include pre-compliance EMC and the safety tests a lab would run, and durability
or life tests where lifetime is a requirement.
## 14. Renders — `11-renders/` (full tier)
- Renders come from the real CAD geometry (the assembly GLB) through Blender in background mode: a product
view, an exploded view, and a view that shows the user interface. Check the scale against a known
dimension. Label them as renders of a digital design.
- Never use generative image models to depict the product: they invent geometry that does not exist.
- You cannot judge images reliably by looking. Check what can be measured (file exists, resolution, the
object fills the frame, parts are present by name) instead of trusting a glance.
## 15. Report and manifest
`REPORT.md` (both tiers) in this order: the product in one paragraph; a key-parameter table (value, unit,
origin label); unit cost at 1/100/1000 and the retail price; status per requirement (from `compliance.csv`);
the top risks and open hazards; decisions taken without a human with their alternatives; what is open,
blocked or not run; for the full tier, the file map. Every number in it must match its home.
`manifest.json` exactly as the RUN CONTRACT specifies, including `key_figures` (use `null` for an unknown
figure, never a guess) and `blocked[]`. List every file under `deliverables/`.
Before you finish: re-read `REPORT.md` and `02-concept/concept.md` against the numbers' homes, check that
`compliance.csv` has one row per requirement id, that every evidence path exists, and that `manifest.json`
parses and lists the files that exist.
## 16. Claims
The project designs real products for honest crowdfunding. In every document: no feature that the design
does not implement, no performance number that is not calculated or sourced, no "certified", "safe",
"waterproof", "AI" or similar word unless the evidence for it is in the deliverables. Under-promise.
## 17. Time and context
- The RUN CONTRACT gives the time target and the hard stop. Plan time boxes per phase at the start, check
`date` at each boundary, and cut scope (depth, not honesty) when behind.
- Long runs lose context. Keep a short state file (decisions, key numbers' homes, open issues, next step)
current, so you or a fresh agent can resume from files alone.
- Write deliverables as soon as each part exists and keep them current.