feat(import): ningen-shikkaku.lifecycle@v2 — import the structured --korg-sink

CHANGELOG.md

@@ -46,6 +46,14 @@ ningen-shikkaku lifecycle metadata (`ningen-shikkaku.lifecycle@v1`), with
 projection-only exporters from Okto session/genome files and `[dazai]` daemon
 logs. Imported events use the new `import:<source>@<version>` actor prefix.
 
+**Added — `ningen-shikkaku.lifecycle@v2` import profile.** Imports the
+structured `--korg-sink` records (`ningen-korg-lifecycle@1`) that `dazai daemon`
+and `motokano` write, instead of scraping `[dazai]` log lines. It covers motokano
+reads and carries no PIDs, tool names, lengths or digests. Exporter
+`ningen_sink_to_envelope` and CLI `python -m korg_ledger envelope ningen-sink
+FILE`, which takes the source version from the records. Spec examples include
+real 0.1.0 sink output.
+
 **Changed.** `POST /api/agent/tool-call` answers **422** for a `triggered_by`
 that is not an existing, strictly-earlier `seq_id` (it used to persist it).
 `LedgerWriter` rejects non-`int` `triggered_by`. The duplicate vector set in

adapters/korg-ledger-py/README.md

@@ -13,6 +13,7 @@ or written (opt out only with `redaction=None`).
     python -m korg_ledger lint ledger.jsonl                 # chain + causal DAG (flat & nested) + redaction
     python -m korg_ledger envelope okto-session S.json --source-version 0.1.0 > env.json
     python -m korg_ledger envelope dazai-log daemon.log --source-version 0.1.0 > env.json
+    python -m korg_ledger envelope ningen-sink lifecycle.jsonl > env.json    # dazai/motokano --korg-sink
     python -m korg_ledger import env.json --ledger ledger.jsonl   # korg-import@v1, atomic on rejection
     python -m korg_ledger rechain-redacted old.jsonl new.jsonl    # migrate a journal that leaked a secret
 

adapters/korg-ledger-py/src/korg_ledger/__main__.py

@@ -3,6 +3,7 @@
     python -m korg_ledger envelope okto-session SESSION.json --source-version 0.1.0 > env.json
     python -m korg_ledger envelope okto-genome genome.jsonl --source-version 0.1.0 > env.json
     python -m korg_ledger envelope dazai-log daemon.log --source-version 0.1.0 > env.json
+    python -m korg_ledger envelope ningen-sink lifecycle.jsonl > env.json   # dazai/motokano --korg-sink
     python -m korg_ledger import env.json --ledger .korg/ledger.jsonl [--triggered-by N]
     python -m korg_ledger lint .korg/ledger.jsonl        # chain + causal DAG + redaction
     python -m korg_ledger rechain-redacted old.jsonl new.jsonl   # migration: redacted copy, new chain
@@ -23,6 +24,7 @@
     ImportRejected,
     dazai_log_to_envelope,
     import_envelope,
+    ningen_sink_to_envelope,
     okto_mutation_record,
     okto_session_to_envelope,
 )
@@ -44,7 +46,16 @@ def _json_objects(text: str) -> list:
 
 def _cmd_envelope(a) -> int:
     text = Path(a.file).read_text(encoding="utf-8")
-    if a.source == "okto-session":
+    if a.source_version is None and a.source != "ningen-sink":
+        print(f"--source-version is required for {a.source}", file=sys.stderr)
+        return 2
+    if a.source == "ningen-sink":
+        try:
+            env = ningen_sink_to_envelope(text.splitlines(), version=a.source_version)
+        except ValueError as ex:
+            print(f"error: {ex} (--source-version)", file=sys.stderr)
+            return 2
+    elif a.source == "okto-session":
         objs = _json_objects(text)
         if len(objs) != 1 or not isinstance(objs[0], dict):
             print("expected exactly one Okto Session object", file=sys.stderr)
@@ -57,7 +68,7 @@ def _cmd_envelope(a) -> int:
             "source": {"name": "okto", "version": a.source_version},
             "records": [okto_mutation_record(m) for m in _json_objects(text) if isinstance(m, dict)],
         }
-    else:
+    elif a.source == "dazai-log":
         env = dazai_log_to_envelope(text.splitlines(), version=a.source_version)
     json.dump(env, sys.stdout, indent=2, sort_keys=True)
     sys.stdout.write("\n")
@@ -116,9 +127,12 @@ def main(argv=None) -> int:
     sub = p.add_subparsers(dest="cmd", required=True)
 
     e = sub.add_parser("envelope", help="build a korg-import@v1 envelope from a sibling tool's native output")
-    e.add_argument("source", choices=["okto-session", "okto-genome", "dazai-log"])
+    e.add_argument("source", choices=["okto-session", "okto-genome", "dazai-log", "ningen-sink"])
     e.add_argument("file")
-    e.add_argument("--source-version", required=True, help="version of the tool that produced FILE")
+    e.add_argument(
+        "--source-version",
+        help="version of the tool that produced FILE (required, except ningen-sink records carry it)",
+    )
     e.set_defaults(fn=_cmd_envelope)
 
     i = sub.add_parser("import", help="validate an envelope and append it to a ledger (atomic on rejection)")

adapters/korg-ledger-py/src/korg_ledger/imports.py

@@ -1,13 +1,17 @@
 """korg-import@v1 — import envelopes for evidence korg did not observe itself.
 
-Normative text: spec/korg-ledger-v1/IMPORT.md. Two profiles ship:
+Normative text: spec/korg-ledger-v1/IMPORT.md. Three profiles ship:
 
 * ``okto.diagnostics@v1`` — sanitized diagnostics from the Okto agent-traffic
   proxy (session/turn/anomaly/mutation/replay *metadata*; never headers, bodies,
   raw stream bytes, or anomaly text).
 * ``ningen-shikkaku.lifecycle@v1`` — lifecycle metadata from the
   ningen-shikkaku dead-man's switch (armed/trigger/panic/wipe/pid events; never a
   secret value, tool spec, argv, or free-text message).
+* ``ningen-shikkaku.lifecycle@v2`` — the structured ``--korg-sink`` records
+  (``ningen-korg-lifecycle@1``) written by ``dazai daemon`` and ``motokano``:
+  session/arm/read/wipe/kill steps keyed by a random per-run session id; no
+  PIDs, tool names, values, lengths or digests.
 
 Safety is by **allowlist**, not by scrubbing: every record kind declares the only
 fields it may carry and their types; anything else rejects the WHOLE envelope
@@ -16,7 +20,7 @@
 the korg-redact@v1 scanner runs over every record as a second gate.
 
 The converters (``okto_session_to_envelope``, ``okto_mutation_record``,
-``dazai_log_to_envelope``) build envelopes by *projection*: they read only the
+``dazai_log_to_envelope``, ``ningen_sink_to_envelope``) build envelopes by *projection*: they read only the
 allowlisted fields from the source's native output, so secret-bearing fields are
 never even copied.
 """
@@ -107,6 +111,17 @@ def _slug_count_map(v, _):
 
 OKTO_PROFILE = "okto.diagnostics@v1"
 NINGEN_PROFILE = "ningen-shikkaku.lifecycle@v1"
+NINGEN_SINK_PROFILE = "ningen-shikkaku.lifecycle@v2"
+
+# ningen-shikkaku.lifecycle@v2: every record names its emitting process and the
+# random per-run session id from the producer's sink; the vocabularies mirror
+# ningen-korg-lifecycle@1 (kiroku) exactly.
+_NINGEN_SINK_COMMON = {"component": _enum("dazai", "motokano"), "session": _id}
+_NINGEN_SINK_REASON = _enum(
+    "heartbeat_lost", "ping_timeout", "panic_signal", "hard_panic_signal", "shutdown",
+    "client_quit", "calls_exhausted", "client_disconnect", "daemon_lost",
+)
+_U32 = _int_range(0, 2**32 - 1)
 
 _AT = {"at_ms": _nonneg_int}
 
@@ -192,6 +207,33 @@ def _slug_count_map(v, _):
             "buffer_address", "address",
         ),
     },
+    NINGEN_SINK_PROFILE: {
+        "source": "ningen-shikkaku",
+        "kinds": {
+            "session_start": ({**_NINGEN_SINK_COMMON, "armed": _bool}, dict(_AT)),
+            "arm": ({**_NINGEN_SINK_COMMON, "via": _enum("startup", "control")}, dict(_AT)),
+            "read": (
+                {**_NINGEN_SINK_COMMON, "tool": _U32, "tool_kind": _enum("static", "exec"),
+                 "served": _bool},
+                dict(_AT),
+            ),
+            "wipe": ({**_NINGEN_SINK_COMMON, "reason": _NINGEN_SINK_REASON, "buffers": _U32}, dict(_AT)),
+            "kill": (
+                {**_NINGEN_SINK_COMMON, "reason": _NINGEN_SINK_REASON,
+                 "target": _enum("registered", "self"), "count": _U32, "executed": _bool},
+                dict(_AT),
+            ),
+            "session_end": ({**_NINGEN_SINK_COMMON, "reason": _NINGEN_SINK_REASON}, dict(_AT)),
+        },
+        # v1's list, plus what v2 deliberately leaves behind: process ids, tool
+        # names, paths, and any length or digest of a secret (a guessing oracle).
+        "forbidden": (
+            "value", "secret", "secret_value", "plaintext", "data", "bytes", "cmd", "command", "spec",
+            "tool_spec", "stdout", "stderr", "output", "argv", "args", "env", "message", "reply",
+            "buffer_address", "address", "pid", "pids", "name", "tool_name", "path", "socket",
+            "length", "len", "size", "hash", "sha256", "digest",
+        ),
+    },
 }
 
 _ENVELOPE_FIELDS = {"schema", "profile", "source", "records", "exported_at_ms", "dropped"}
@@ -496,5 +538,84 @@ def dazai_log_to_envelope(lines: Iterable[str], *, version: str) -> dict:
     return env
 
 
+NINGEN_SINK_CONTRACT = "ningen-korg-lifecycle@1"
+
+# sink tool_name → (v2 kind, [(v2 field, "args"|"result", sink field)]).
+# Projection reads ONLY these fields; component/session/at_ms are added for all.
+_NINGEN_SINK_STEPS: dict[str, tuple[str, list[tuple[str, str, str]]]] = {
+    "ningen.session_start": ("session_start", [("armed", "args", "armed")]),
+    "ningen.arm": ("arm", [("via", "args", "via")]),
+    "ningen.read": ("read", [("tool", "args", "tool"), ("tool_kind", "args", "kind"),
+                             ("served", "result", "served")]),
+    "ningen.wipe": ("wipe", [("reason", "args", "reason"), ("buffers", "args", "buffers")]),
+    "ningen.kill": ("kill", [("reason", "args", "reason"), ("target", "args", "target"),
+                             ("count", "args", "count"), ("executed", "result", "executed")]),
+    "ningen.session_end": ("session_end", [("reason", "args", "reason")]),
+}
+
+
+def ningen_sink_to_envelope(lines: Iterable[str], *, version: str | None = None) -> dict:
+    """Project a ningen-shikkaku ``--korg-sink`` file (``ningen-korg-lifecycle@1``
+    JSONL from ``dazai daemon`` / ``motokano``) onto
+    ``ningen-shikkaku.lifecycle@v2``.
+
+    Each line is read only through the fixed field map above; any other field a
+    line carries is never copied. Lines that are not JSON objects of that
+    contract (a torn final line after a crash, foreign records) are dropped and
+    counted in ``dropped``. Values are not coerced: a projected value outside the
+    profile's vocabulary makes the import reject the whole envelope.
+
+    ``version`` defaults to the producer version stamped in every record
+    (``args.producer = "ningen-shikkaku@<version>"``); a sink spanning several
+    producer versions needs an explicit ``version``."""
+    records: list[dict] = []
+    producers: set[Any] = set()
+    dropped = 0
+    for line in lines:
+        if not line.strip():
+            continue
+        try:
+            obj = json.loads(line)
+        except json.JSONDecodeError:
+            dropped += 1
+            continue
+        args = obj.get("args") if isinstance(obj, dict) else None
+        result = obj.get("result") if isinstance(obj, dict) else None
+        step = _NINGEN_SINK_STEPS.get(obj.get("tool_name")) if isinstance(obj, dict) else None
+        if (
+            step is None
+            or not isinstance(args, dict)
+            or not isinstance(result, dict)
+            or args.get("contract") != NINGEN_SINK_CONTRACT
+        ):
+            dropped += 1
+            continue
+        kind, fields = step
+        rec: dict[str, Any] = {"kind": kind}
+        # An absent source field stays absent, so the import names it missing.
+        for dst, where, src in [("component", "args", "component"), ("session", "args", "session"), *fields]:
+            source = args if where == "args" else result
+            if src in source:
+                rec[dst] = source[src]
+        at = args.get("observed_unix_ms")
+        if _nonneg_int(at, None):
+            rec["at_ms"] = at
+        records.append(rec)
+        producers.add(args.get("producer"))
+    if version is None:
+        prefix = "ningen-shikkaku@"
+        found = sorted(p[len(prefix):] for p in producers if isinstance(p, str) and p.startswith(prefix))
+        if len(found) != 1 or len(producers) != 1:
+            raise ValueError(
+                "sink records do not name exactly one ningen-shikkaku producer version; pass version explicitly"
+            )
+        version = found[0]
+    env = {"schema": SCHEMA, "profile": NINGEN_SINK_PROFILE,
+           "source": {"name": "ningen-shikkaku", "version": version}, "records": records}
+    if dropped:
+        env["dropped"] = dropped
+    return env
+
+
 def load_envelope(text: str) -> dict:
     return json.loads(text)

adapters/korg-ledger-py/tests/test_imports.py

@@ -11,6 +11,7 @@
     dazai_log_to_envelope,
     envelope_digest,
     import_envelope,
+    ningen_sink_to_envelope,
     okto_mutation_record,
     okto_session_to_envelope,
     validate_envelope,
@@ -28,7 +29,14 @@ def _events(path):
     return [json.loads(l) for l in Path(path).read_text().splitlines() if l.strip()]
 
 
-@pytest.mark.parametrize("name", ["okto-diagnostics.example.json", "ningen-shikkaku-lifecycle.example.json"])
+@pytest.mark.parametrize(
+    "name",
+    [
+        "okto-diagnostics.example.json",
+        "ningen-shikkaku-lifecycle.example.json",
+        "ningen-shikkaku-lifecycle-v2.example.json",
+    ],
+)
 def test_spec_examples_import_into_a_verifiable_ledger(tmp_path, name):
     env = _load(name)
     assert validate_envelope(env) == []
@@ -60,6 +68,11 @@ def test_spec_examples_import_into_a_verifiable_ledger(tmp_path, name):
             "records[0]: forbidden field 'value' (may carry secrets or raw traffic)",
             "records[1]: field 'reason' has an invalid value",
         ]),
+        ("ningen-shikkaku-lifecycle-v2.rejected.json", [
+            "records[0]: forbidden field 'name' (may carry secrets or raw traffic)",
+            "records[1]: field 'reason' has an invalid value",
+            "records[1]: forbidden field 'pid' (may carry secrets or raw traffic)",
+        ]),
     ],
 )
 def test_spec_rejections_are_atomic(tmp_path, name, expected):
@@ -215,3 +228,103 @@ def test_cli_round_trip(tmp_path, capsys):
     before = led.read_text()
     assert cli(["import", str(bad), "--ledger", str(led)]) == 1
     assert led.read_text() == before
+
+
+# ── ningen-shikkaku.lifecycle@v2: the structured --korg-sink ─────────────────
+
+SINK_LINES = (IMPORTS / "ningen-shikkaku-sink.example.jsonl").read_text(encoding="utf-8").splitlines()
+
+
+def test_ningen_sink_projection_reproduces_the_spec_example():
+    # The sink example is real output of `dazai daemon` + `motokano` 0.1.0.
+    env = ningen_sink_to_envelope(SINK_LINES)
+    assert env == _load("ningen-shikkaku-lifecycle-v2.example.json")
+    assert [r["kind"] for r in env["records"]] == [
+        "session_start", "arm", "kill", "wipe", "kill",
+        "session_start", "read", "read", "read", "wipe", "session_end",
+    ]
+    assert "dropped" not in env
+
+
+def test_ningen_sink_projection_never_copies_unlisted_fields_and_drops_junk(tmp_path):
+    # A drifted or hostile producer: extra secret-bearing fields on otherwise
+    # valid lines, a torn final line, a foreign record, a different contract.
+    tainted = []
+    for line in SINK_LINES:
+        rec = json.loads(line)
+        rec["secret"] = SECRET
+        rec["args"].update(value=SECRET, name="get_key", pid=4242, cmd=f"echo {SECRET}")
+        rec["result"].update(stdout=SECRET, len=len(SECRET))
+        tainted.append(json.dumps(rec))
+    foreign = json.loads(SINK_LINES[0])
+    foreign["args"]["contract"] = "something-else@1"
+    lines = tainted + [
+        json.dumps(foreign),
+        json.dumps({"tool_name": "Bash", "args": {"command": f"export TOKEN={SECRET}"}, "result": {}}),
+        "[dazai] error: not json",
+        SINK_LINES[0][:40],  # torn by a crash mid-write
+    ]
+    env = ningen_sink_to_envelope(lines)
+    blob = json.dumps(env)
+    for leaked in (SECRET, "get_key", "4242", "echo", "stdout"):
+        assert leaked not in blob
+    assert env["dropped"] == 4
+    assert env["records"] == _load("ningen-shikkaku-lifecycle-v2.example.json")["records"]
+    led = tmp_path / "l.jsonl"
+    import_envelope(LedgerWriter(led), env)
+    assert SECRET not in led.read_text() and verify_redaction(_events(led)) == []
+
+
+@pytest.mark.parametrize(
+    "mutate, needle",
+    [
+        (lambda a, r: a.update(reason="heartbeat connection dropped"), "'reason' has an invalid value"),
+        (lambda a, r: a.update(component="rei"), "'component' has an invalid value"),
+        (lambda a, r: a.update(session="has spaces"), "'session' has an invalid value"),
+        (lambda a, r: a.pop("session"), "missing required field 'session'"),
+        (lambda a, r: a.update(count=-1), "'count' has an invalid value"),
+        (lambda a, r: a.update(count=2**32), "'count' has an invalid value"),
+        (lambda a, r: r.update(executed="yes"), "'executed' has an invalid value"),
+    ],
+)
+def test_ningen_sink_values_outside_the_vocabulary_reject_the_whole_envelope(tmp_path, mutate, needle):
+    kill = json.loads(SINK_LINES[2])
+    assert kill["tool_name"] == "ningen.kill"
+    mutate(kill["args"], kill["result"])
+    env = ningen_sink_to_envelope(SINK_LINES[:2] + [json.dumps(kill)] + SINK_LINES[3:])
+    errors = validate_envelope(env)
+    assert any(e.startswith("records[2]: ") and needle in e for e in errors), errors
+    led = tmp_path / "l.jsonl"
+    with pytest.raises(ImportRejected):
+        import_envelope(LedgerWriter(led), env)
+    assert led.read_text() == ""
+
+
+def test_ningen_sink_version_comes_from_the_producer_unless_ambiguous():
+    assert ningen_sink_to_envelope(SINK_LINES)["source"]["version"] == "0.1.0"
+    assert ningen_sink_to_envelope(SINK_LINES, version="9.9.9")["source"]["version"] == "9.9.9"
+    newer = json.loads(SINK_LINES[0])
+    newer["args"]["producer"] = "ningen-shikkaku@0.2.0"
+    with pytest.raises(ValueError, match="pass version explicitly"):
+        ningen_sink_to_envelope(SINK_LINES + [json.dumps(newer)])
+    assert ningen_sink_to_envelope(SINK_LINES + [json.dumps(newer)], version="0.2.0")["records"]
+
+
+def test_ningen_sink_cli_round_trip(tmp_path, capsys):
+    sink = tmp_path / "lifecycle.jsonl"
+    sink.write_text("\n".join(SINK_LINES) + "\n")
+    assert cli(["envelope", "ningen-sink", str(sink)]) == 0
+    env_path = tmp_path / "env.json"
+    env_path.write_text(capsys.readouterr().out)
+    led = tmp_path / "ledger.jsonl"
+    assert cli(["import", str(env_path), "--ledger", str(led)]) == 0
+    assert cli(["lint", str(led)]) == 0
+    events = _events(led)
+    assert [e["event"]["tool_name"] for e in events[:3]] == [
+        "korg.import", "ningen-shikkaku.session_start", "ningen-shikkaku.arm",
+    ]
+    assert {e["event"]["source_agent"] for e in events} == {"import:ningen-shikkaku@0.1.0"}
+    # Other sources still need an explicit version.
+    log = tmp_path / "daemon.log"
+    log.write_text(DAZAI_LOG)
+    assert cli(["envelope", "dazai-log", str(log)]) == 2