Apply this competence to the work described below. The public edition 0.3.2 includes its method, operating knowledge and worked examples. Add your context after the supplied knowledge. Use execution tools only if this host provides them. # D-ND Material Work Continuity Preserve original requests, actions, corrections and the present position without rewriting what happened. Keep material events connected and distinguish the person's source from assistant interpretations and observed effects. Use this competence when requirements or direction change, or when a project’s present state needs causal reconstruction. A decision can change without erasing earlier events. Connect original request, AI interpretation, actual work and later correction. Keep detailed history reachable while the current entry contains what is needed to continue. ## Preserve the source before its account When a request, decision or correction materially changes work, preserve its original words or an exact accessible reference. Keep uncertainty and unusual wording. Attribute the assistant's summary and inferences separately. If the source is absent, identify what cannot be reconstructed rather than recreating a quotation from memory or a summary. Use the existing continuity location when suitable. ## Follow events and effects Record events that change reconstruction: a source, AI proposal, evidenced action, correction, authorization change or decisive unknown. Connect each effect to its evidence and observation time. Preparing a file, sending it and receiving a reply are different facts. Retain the action that occurred even when its producing plan is later abandoned. ## Correct without rewriting the past A correction adds a relation to history: identify the interpretation or instruction superseded and the continuation changed. Do not erase the previous event or backdate later understanding. Revoking permission changes what can happen now without undoing an earlier effect. An AI summary cannot create authorization; retain the person's or responsible party's actual decision reference. ## Make the present reconstructable Keep a short entry connecting still-valid sources, decisions, completed effects, unknowns and the pertinent next movement. Detailed register and current entry have different functions. Use identified events and predecessor references when complexity requires them; attributed dated notes may otherwise suffice. At reentry, check that useful references are reachable and identify absent or changed material. Do not claim coverage of an unread source. ## Preserve work and improve reconstruction Write within the selected project while preserving concurrent contributions. If persistence is unavailable, prepare a note the person can retain. Do not build a universal diary. Exclude unrelated private data and secrets, including derived fingerprints. When reentry exposes causal loss, improve the continuity method so the next note supports better reconstruction rather than adding volume alone. ## Method depth This competence concerns material continuity: what happened and the sources it derives from. It does not alone determine a decision's meaning or the whole project's state. The relation becomes meta when reconstruction failure changes preservation itself. A summary may have mixed proposal and approval; continuation makes both recognizable. This is distinct from autonomous execution, and a well-formed register does not establish infallible memory. ## Learning and continuation A handoff reveals which connections suffice and which are missing. Retain the correction in continuity while leaving project choices with the responsible party. Begin with available context and sources. The method can operate in chat; persistence, tools and external actions depend on the recipient's actual means and mandate. Instructions make a capability usable; its behavior and value are observed in real work. ## Reachable operating knowledge Read [OPERATING_GUIDE.md](OPERATING_GUIDE.md) as part of this method when forming the work. It carries the practical knowledge, decision relations and available means for this capability. The complete chat starter includes this knowledge and worked examples, so it remains usable without private sources. Preserve the public source identity and adapt through the recipient's actual means. # D-ND Material Work Continuity ## Use the deterministic recorder when exact event integrity matters The textual method is usable on its own. This kit also supplies the source competence's Python reducer, event/state JSON schema, a worked ledger and portable tests. The reducer is a record-integrity tool, not a semantic admission gate or an independent authority. Hashes detect inconsistent records; they do not prove that a statement or evidence reference is true. Keep the original ledger and source locators in the project that owns them. The seven event kinds preserve exact source, assistant projection, observed movement, retraction or supersession, authority change, unknown change and a current-relation checkpoint. Sources retain attribution. Projections cite preceding events. A correction supersedes an interpretation without deleting its event or erasing observed effects. A checkpoint exposes current sources, remaining unknowns and the next permitted action. Secret-classified sources retain a gap marker, without secret values or hashes. Never copy a credential into a record. ## Run the included example Use Python 3.10 or later from the competence folder. Run `python scripts/demo_continuity.py --output-dir demo-ledger` with a new directory name. It writes an initial state, six illustrative events and their successor states, and `reconstruction.json`. The example corrects an assistant's publication interpretation while preserving a draft actually saved. It does not publish or contact anything. Run `python scripts/reduce_material_work_continuity.py verify --state demo-ledger/C1.state.json` to reconstruct that state. For your project, `init` accepts `--owner-surface`, `--owner-state-ref`, `--authority` and optional `--output`. `reduce` accepts `--state`, `--event` and optional `--output`; use a separate output file to retain the preceding state. An event's `previous_state_digest` must match the state it extends. The demo shows how to construct each event through the Python interface. ## Understand failures and limits An exact replay is idempotent; an altered event with the same identity is rejected. A stale predecessor, tampered state, fabricated authority reference or unsupported resolution of an unknown is rejected. A structured authority-grant source models a recorded change; the tool cannot authenticate the human behind an externally supplied document. A referenced source must be an absolute regular file with the recorded digest, at most 1 MiB; missing, symlinked or changed locators produce incomplete reconstruction. Inline exact sources remain available in the ledger. The CLI writes only its chosen output and has no network calls. Concurrent writers need serialization at the actual file/store boundary; this reducer alone supplies no file lock or database transaction. A schema-valid state is also checked by replaying its event chain. Run the portable tests with `python -m pip install -r requirements-test.txt` followed by `python scripts/test_reduce_material_work_continuity.py`. The reducer itself uses only the standard library; `jsonschema` is needed by schema tests. The schema retains original protocol names for compatibility, with a public schema identifier and Windows/POSIX source paths. # D-ND Material Work Continuity — application examples [Italiano](ESEMPI_IT.md). The following cases illustrate how the method is used. ## Preserve an earlier approval when the client changes direction A client requests one page and approves a draft, then selects two separate paths. Preserve the original request and approval, the draft produced and the new decision; identify work that remains useful. The current entry leads to the new structure without making the earlier approval disappear or appear mistaken. ## Keep a sent version distinct from its later correction A document was sent before a figure was corrected. Record the identity of the sent version and the observed effect, then add the correction. An updated local copy does not replace the document already received. Any subsequent sending follows the actual mandate. ## Reconstruct a project in a new assistant The new chat receives sources, decisions and documented actions. A cited conversation is no longer accessible: identify that gap and continue from available information. Reconstruction distinguishes what can be checked from what is known only through a summary. ## Apply the method to your case Use [SKILL.md](SKILL.md) or [START_CHAT_EN.txt](START_CHAT_EN.txt) with the relevant sources and your actual context. Keep observations, interpretations and what remains open distinguishable. Preserve the useful reason and continuation in the project’s guide. # Included executable sources ## scripts/reduce_material_work_continuity.py ```python #!/usr/bin/env python3 """Deterministic append-only reducer for material D-ND work continuity.""" from __future__ import annotations import argparse import copy import hashlib import json from pathlib import Path from typing import Any STATE_SCHEMA = "dnd.material_work_continuity.state.v1" EVENT_SCHEMA = "dnd.material_work_continuity.event.v1" EVENT_KINDS = { "source_event", "assistant_projection", "verified_movement", "retraction_or_supersession", "authority_change", "unknown_change", "relation_checkpoint", } SOURCE_PROVENANCE = { "direct_operator", "owner_source", "operator_submitted_uncertain_authorship", } STATE_KEYS = { "schema", "version", "owner_surface", "owner_state_ref", "authority_initial", "authority_current", "events", "active_event_ids", "superseded_event_ids", "unknowns", "current_relation", "next_legal_action", "reopen_condition", "finite_state", "previous_digest", "state_digest", } EVENT_KEYS = { "schema", "event_id", "event_kind", "owner_surface", "previous_state_digest", "payload", } PAYLOAD_KEYS = { "source_event": { "provenance_class", "source_ref", "exact_text", "source_digest", "secret_classified", "source_gap_marker", "operative_status", }, "assistant_projection": { "projection", "derives_from", "operative_status", }, "verified_movement": { "action_requested", "action_performed", "evidence_refs", "effects_observed", "authority_before", "authority_after", }, "retraction_or_supersession": { "retraction_authority_ref", "retracts", "scope", "effective_relation_change", "effects_not_reversed", }, "authority_change": { "authority_before", "authority_after", "owner_receipt_ref", }, "unknown_change": {"adds", "resolves", "evidence_refs"}, "relation_checkpoint": { "source_event_refs", "projection_refs", "current_relation", "unknowns", "next_legal_action", "reopen_condition", }, } class ContinuityError(ValueError): """Closed reducer failure with a stable reason.""" def canonical_json(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def sha256_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def _state_digest(state: dict[str, Any]) -> str: body = copy.deepcopy(state) body.pop("state_digest", None) return sha256_text(canonical_json(body)) def _seal(state: dict[str, Any]) -> dict[str, Any]: sealed = copy.deepcopy(state) sealed["state_digest"] = _state_digest(sealed) return sealed def _require(condition: bool, reason: str) -> None: if not condition: raise ContinuityError(reason) def _require_string(value: Any, reason: str, *, allow_blank: bool = False) -> str: _require(isinstance(value, str), reason) if not allow_blank: _require(bool(value.strip()), reason) return value def _require_string_list(value: Any, reason: str) -> list[str]: _require(isinstance(value, list), reason) _require(all(isinstance(item, str) and item.strip() for item in value), reason) _require(len(value) == len(set(value)), reason) return list(value) def initial_state(owner_surface: str, owner_state_ref: str, authority: str) -> dict[str, Any]: for value, label in ( (owner_surface, "owner_surface required"), (owner_state_ref, "owner_state_ref required"), (authority, "authority required"), ): _require_string(value, label) return _seal( { "schema": STATE_SCHEMA, "version": 1, "owner_surface": owner_surface, "owner_state_ref": owner_state_ref, "authority_initial": authority, "authority_current": authority, "events": [], "active_event_ids": [], "superseded_event_ids": [], "unknowns": [], "current_relation": "", "next_legal_action": "", "reopen_condition": "", "finite_state": "EMPTY_CONTINUITY", "previous_digest": None, "state_digest": "", } ) def _basic_state_validation(state: dict[str, Any]) -> None: _require(isinstance(state, dict), "state must be an object") _require(set(state) == STATE_KEYS, "state fields are not closed") _require(state["schema"] == STATE_SCHEMA, "unsupported state schema") _require(state["version"] == 1, "unsupported state version") _require_string(state["owner_surface"], "owner_surface required") _require_string(state["owner_state_ref"], "owner_state_ref required") _require_string(state["authority_initial"], "authority_initial required") _require_string(state["authority_current"], "authority_current required") _require(isinstance(state["events"], list), "events must be a list") event_ids = [event.get("event_id") for event in state["events"] if isinstance(event, dict)] _require(len(event_ids) == len(state["events"]), "every event must be an object with event_id") _require(len(event_ids) == len(set(event_ids)), "duplicate event IDs in state") _require_string_list(state["active_event_ids"], "active_event_ids invalid") _require_string_list(state["superseded_event_ids"], "superseded_event_ids invalid") _require_string_list(state["unknowns"], "unknowns invalid") _require(set(state["active_event_ids"]).isdisjoint(state["superseded_event_ids"]), "event cannot be active and superseded") _require(set(state["active_event_ids"]) | set(state["superseded_event_ids"]) == set(event_ids), "event status coverage mismatch") _require_string(state["current_relation"], "current_relation invalid", allow_blank=True) _require_string(state["next_legal_action"], "next_legal_action invalid", allow_blank=True) _require_string(state["reopen_condition"], "reopen_condition invalid", allow_blank=True) _require(state["finite_state"] in { "EMPTY_CONTINUITY", "CONTINUITY_CAPTURED", "CURRENT_RELATION_RECONSTRUCTABLE", "INCOMPLETE_SOURCE_RECONSTRUCTION", }, "invalid reducer finite state") _require(state["previous_digest"] is None or isinstance(state["previous_digest"], str), "previous_digest invalid") _require(isinstance(state["state_digest"], str) and len(state["state_digest"]) == 64, "state_digest invalid") _require(_state_digest(state) == state["state_digest"], "state digest mismatch") def _event_index(state: dict[str, Any]) -> dict[str, dict[str, Any]]: return {event["event_id"]: event for event in state["events"]} def _validate_event_shape(state: dict[str, Any], event: dict[str, Any]) -> None: _require(isinstance(event, dict), "event must be an object") _require(set(event) == EVENT_KEYS, "event fields are not closed") _require(event["schema"] == EVENT_SCHEMA, "unsupported event schema") _require_string(event["event_id"], "event_id required") _require(event["event_kind"] in EVENT_KINDS, "unsupported event kind") _require(event["owner_surface"] == state["owner_surface"], "event owner does not match state owner") _require_string(event["previous_state_digest"], "previous_state_digest required") _require(isinstance(event["payload"], dict), "event payload must be an object") _require(set(event["payload"]) == PAYLOAD_KEYS[event["event_kind"]], "event payload fields are not closed") def _validate_source_event(payload: dict[str, Any]) -> None: _require(payload["provenance_class"] in SOURCE_PROVENANCE, "invalid source provenance") _require_string(payload["source_ref"], "source_ref required") _require(isinstance(payload["secret_classified"], bool), "secret_classified must be boolean") _require(payload["operative_status"] == "active", "new source event must be active") if payload["secret_classified"]: _require(payload["exact_text"] == "", "secret source exact text is forbidden") _require(payload["source_digest"] == "", "secret source digest is forbidden") _require_string(payload["source_gap_marker"], "secret source gap marker required") else: _require(payload["source_gap_marker"] == "", "non-secret source cannot carry a secret gap marker") _require_string(payload["source_digest"], "source digest required") if payload["exact_text"]: _require(sha256_text(payload["exact_text"]) == payload["source_digest"], "source exact text digest mismatch") else: _require(Path(payload["source_ref"]).is_absolute(), "external exact source_ref must be absolute") def _source_text(payload: dict[str, Any]) -> str: if payload["exact_text"]: return payload["exact_text"] path = Path(payload["source_ref"]) _require(path.is_file() and not path.is_symlink(), "owner receipt source unavailable") _require(path.stat().st_size <= 1024 * 1024, "owner receipt source too large") raw = path.read_bytes() _require(hashlib.sha256(raw).hexdigest() == payload["source_digest"], "owner receipt source digest mismatch") try: return raw.decode("utf-8") except UnicodeDecodeError as exc: raise ContinuityError("owner receipt source is not UTF-8") from exc def _validate_authority_grant(payload: dict[str, Any], before: str, after: str) -> None: try: grant = json.loads(_source_text(payload)) except json.JSONDecodeError as exc: raise ContinuityError("owner receipt is not a structured authority grant") from exc _require(isinstance(grant, dict), "owner receipt is not a structured authority grant") _require( set(grant) == {"schema", "authority_before", "authority_after"}, "owner receipt authority grant fields are not closed", ) _require( grant["schema"] == "dnd.material_work_continuity.authority_grant.v1", "owner receipt authority grant schema invalid", ) _require(grant["authority_before"] == before, "owner receipt authority_before mismatch") _require(grant["authority_after"] == after, "owner receipt authority_after mismatch") def _validate_payload(state: dict[str, Any], event: dict[str, Any]) -> None: kind = event["event_kind"] payload = event["payload"] index = _event_index(state) if kind == "source_event": _validate_source_event(payload) elif kind == "assistant_projection": _require_string(payload["projection"], "projection required") refs = _require_string_list(payload["derives_from"], "derives_from invalid") _require(refs and all(ref in index for ref in refs), "projection derivation ref missing") _require(payload["operative_status"] == "active", "new projection must be active") elif kind == "verified_movement": _require_string(payload["action_requested"], "action_requested required") _require_string(payload["action_performed"], "action_performed required") _require_string_list(payload["evidence_refs"], "evidence_refs invalid") _require(bool(payload["evidence_refs"]), "verified movement requires evidence") _require_string_list(payload["effects_observed"], "effects_observed invalid") _require_string(payload["authority_before"], "authority_before required") _require_string(payload["authority_after"], "authority_after required") _require(payload["authority_before"] == state["authority_current"], "movement authority_before mismatch") _require(payload["authority_after"] == payload["authority_before"], "authority changes require authority_change event") elif kind == "retraction_or_supersession": authority_ref = _require_string(payload["retraction_authority_ref"], "retraction authority ref required") targets = _require_string_list(payload["retracts"], "retracts invalid") _require(bool(targets), "retraction target required") _require(authority_ref in index, "retraction authority event missing") authority_event = index[authority_ref] _require(authority_event["event_kind"] == "source_event", "retraction authority must be source event") _require(authority_event["payload"]["provenance_class"] in {"direct_operator", "owner_source"}, "retraction authority must be owner source") _require(authority_ref in state["active_event_ids"], "retraction authority source is not active") _require(authority_ref not in targets, "retraction cannot retract its own authority") _require(all(target in index for target in targets), "retraction target missing") _require(all(target not in state["superseded_event_ids"] for target in targets), "target already superseded") _require_string(payload["scope"], "retraction scope required") _require_string(payload["effective_relation_change"], "effective relation change required") effects_not_reversed = _require_string_list(payload["effects_not_reversed"], "effects_not_reversed invalid") required_effects = { effect for target in targets if index[target]["event_kind"] == "verified_movement" for effect in index[target]["payload"]["effects_observed"] } _require(required_effects <= set(effects_not_reversed), "retraction erased an observed effect") elif kind == "authority_change": _require_string(payload["authority_before"], "authority_before required") _require_string(payload["authority_after"], "authority_after required") owner_receipt_ref = _require_string(payload["owner_receipt_ref"], "owner receipt required") _require(payload["authority_before"] == state["authority_current"], "authority_before mismatch") _require(payload["authority_after"] != payload["authority_before"], "authority change must change authority") _require(owner_receipt_ref in index, "owner receipt event missing") owner_receipt = index[owner_receipt_ref] _require(owner_receipt["event_kind"] == "source_event", "owner receipt must be source event") _require( owner_receipt["payload"]["provenance_class"] in {"direct_operator", "owner_source"}, "owner receipt must be operator or owner source", ) _require(owner_receipt_ref in state["active_event_ids"], "owner receipt is not active") _validate_authority_grant( owner_receipt["payload"], payload["authority_before"], payload["authority_after"], ) elif kind == "unknown_change": adds = _require_string_list(payload["adds"], "unknown adds invalid") resolves = _require_string_list(payload["resolves"], "unknown resolves invalid") evidence = _require_string_list(payload["evidence_refs"], "unknown evidence invalid") _require(set(adds).isdisjoint(resolves), "unknown cannot be added and resolved together") _require(set(resolves) <= set(state["unknowns"]), "resolved unknown not present") if resolves: _require(bool(evidence), "unknown resolution requires evidence") elif kind == "relation_checkpoint": source_refs = _require_string_list(payload["source_event_refs"], "source_event_refs invalid") projection_refs = _require_string_list(payload["projection_refs"], "projection_refs invalid") _require(bool(source_refs), "relation checkpoint requires source") for ref in source_refs: _require(ref in index and index[ref]["event_kind"] == "source_event", "checkpoint source ref invalid") _require(ref not in state["superseded_event_ids"], "checkpoint cannot use superseded source") for ref in projection_refs: _require(ref in index and index[ref]["event_kind"] == "assistant_projection", "checkpoint projection ref invalid") _require(ref not in state["superseded_event_ids"], "checkpoint cannot use superseded projection") _require_string(payload["current_relation"], "current relation required") _require_string_list(payload["unknowns"], "checkpoint unknowns invalid") _require(set(payload["unknowns"]) == set(state["unknowns"]), "checkpoint unknown coverage mismatch") _require_string(payload["next_legal_action"], "next legal action required") _require_string(payload["reopen_condition"], "reopen condition required") def _active_source_gap(state: dict[str, Any]) -> bool: index = _event_index(state) return any( index[event_id]["event_kind"] == "source_event" and index[event_id]["payload"]["secret_classified"] for event_id in state["active_event_ids"] ) def apply_event(state: dict[str, Any], event: dict[str, Any], *, _replay: bool = False) -> dict[str, Any]: verify_state(state, full=not _replay) _validate_event_shape(state, event) index = _event_index(state) existing = index.get(event["event_id"]) if existing is not None: if canonical_json(existing) == canonical_json(event): return copy.deepcopy(state) raise ContinuityError("event ID collision") _require(event["previous_state_digest"] == state["state_digest"], "state head mismatch") _validate_payload(state, event) updated = copy.deepcopy(state) old_digest = state["state_digest"] updated["events"].append(copy.deepcopy(event)) updated["active_event_ids"].append(event["event_id"]) kind = event["event_kind"] payload = event["payload"] if kind == "retraction_or_supersession": for target in payload["retracts"]: updated["active_event_ids"].remove(target) updated["superseded_event_ids"].append(target) elif kind == "authority_change": updated["authority_current"] = payload["authority_after"] elif kind == "unknown_change": updated["unknowns"] = sorted( (set(updated["unknowns"]) | set(payload["adds"])) - set(payload["resolves"]) ) elif kind == "relation_checkpoint": updated["current_relation"] = payload["current_relation"] updated["next_legal_action"] = payload["next_legal_action"] updated["reopen_condition"] = payload["reopen_condition"] if _active_source_gap(updated): updated["finite_state"] = "INCOMPLETE_SOURCE_RECONSTRUCTION" elif kind == "relation_checkpoint": updated["finite_state"] = "CURRENT_RELATION_RECONSTRUCTABLE" else: updated["finite_state"] = "CONTINUITY_CAPTURED" updated["previous_digest"] = old_digest updated["state_digest"] = "" return _seal(updated) def verify_state(state: dict[str, Any], *, full: bool = True) -> None: _basic_state_validation(state) if not full: return rebuilt = initial_state( state["owner_surface"], state["owner_state_ref"], state["authority_initial"], ) for event in state["events"]: rebuilt = apply_event(rebuilt, event, _replay=True) _require(canonical_json(rebuilt) == canonical_json(state), "state event chain mismatch") def verify_source_locators(state: dict[str, Any]) -> list[str]: verify_state(state) failures: list[str] = [] for event in state["events"]: if event["event_kind"] != "source_event": continue payload = event["payload"] if payload["secret_classified"] or payload["exact_text"]: continue path = Path(payload["source_ref"]) if not path.is_file() or path.is_symlink(): failures.append(f"{event['event_id']}:source_unavailable") continue if path.stat().st_size > 1024 * 1024: failures.append(f"{event['event_id']}:source_too_large") continue digest = hashlib.sha256(path.read_bytes()).hexdigest() if digest != payload["source_digest"]: failures.append(f"{event['event_id']}:source_digest_mismatch") return failures def reconstruct(state: dict[str, Any]) -> dict[str, Any]: verify_state(state) locator_failures = verify_source_locators(state) index = _event_index(state) effects = sorted( { effect for event in state["events"] if event["event_kind"] == "verified_movement" for effect in event["payload"]["effects_observed"] } ) active_sources = [ event_id for event_id in state["active_event_ids"] if index[event_id]["event_kind"] == "source_event" ] active_projections = [ event_id for event_id in state["active_event_ids"] if index[event_id]["event_kind"] == "assistant_projection" ] movements = [ event["event_id"] for event in state["events"] if event["event_kind"] == "verified_movement" ] retractions = [ { "event_id": event["event_id"], "retraction_authority_ref": event["payload"]["retraction_authority_ref"], "retracts": list(event["payload"]["retracts"]), "scope": event["payload"]["scope"], "effective_relation_change": event["payload"]["effective_relation_change"], "effects_not_reversed": list(event["payload"]["effects_not_reversed"]), } for event in state["events"] if event["event_kind"] == "retraction_or_supersession" ] active_checkpoints = [ event for event in state["events"] if event["event_kind"] == "relation_checkpoint" and event["event_id"] in state["active_event_ids"] ] latest_checkpoint = active_checkpoints[-1] if active_checkpoints else None finite_state = state["finite_state"] if locator_failures: finite_state = "INCOMPLETE_SOURCE_RECONSTRUCTION" return { "owner_surface": state["owner_surface"], "owner_state_ref": state["owner_state_ref"], "source_event_refs": active_sources, "active_source_event_refs": active_sources, "projection_refs": active_projections, "verified_movement_refs": movements, "retractions": retractions, "superseded_event_refs": list(state["superseded_event_ids"]), "effects_preserved": effects, "authority_before": state["authority_initial"], "authority_after": state["authority_current"], "authority_current": state["authority_current"], "unknowns": list(state["unknowns"]), "checkpoint_ref": latest_checkpoint["event_id"] if latest_checkpoint else None, "checkpoint": copy.deepcopy(latest_checkpoint["payload"]) if latest_checkpoint else None, "current_relation": state["current_relation"], "next_legal_action": state["next_legal_action"], "reopen_condition": state["reopen_condition"], "source_locator_failures": locator_failures, "coverage": { "event_count": len(state["events"]), "active_event_count": len(state["active_event_ids"]), "superseded_event_count": len(state["superseded_event_ids"]), "source_locator_failures": locator_failures, }, "finite_state": finite_state, "state_digest": state["state_digest"], } def _read_json(path: str) -> dict[str, Any]: data = json.loads(Path(path).read_text(encoding="utf-8")) _require(isinstance(data, dict), "JSON root must be an object") return data def _emit(value: dict[str, Any], output: str | None) -> None: rendered = json.dumps(value, indent=2, ensure_ascii=False) + "\n" if output: Path(output).write_text(rendered, encoding="utf-8") else: print(rendered, end="") def main() -> int: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command", required=True) init = sub.add_parser("init") init.add_argument("--owner-surface", required=True) init.add_argument("--owner-state-ref", required=True) init.add_argument("--authority", required=True) init.add_argument("--output") reduce_cmd = sub.add_parser("reduce") reduce_cmd.add_argument("--state", required=True) reduce_cmd.add_argument("--event", required=True) reduce_cmd.add_argument("--output") verify = sub.add_parser("verify") verify.add_argument("--state", required=True) args = parser.parse_args() try: if args.command == "init": value = initial_state(args.owner_surface, args.owner_state_ref, args.authority) elif args.command == "reduce": value = apply_event(_read_json(args.state), _read_json(args.event)) else: value = reconstruct(_read_json(args.state)) except (ContinuityError, OSError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False)) return 2 _emit(value, getattr(args, "output", None)) return 0 if __name__ == "__main__": raise SystemExit(main()) ``` ## scripts/demo_continuity.py ```python """Create a small illustrative ledger in a new directory (Python 3.10+).""" import argparse import hashlib import json from pathlib import Path from reduce_material_work_continuity import EVENT_SCHEMA, apply_event, initial_state, reconstruct def run(output): output.mkdir(parents=True, exist_ok=False) state = initial_state('Example project', 'CURRENT_STATE.md', 'project owner') def save(name, value): (output / name).write_text(json.dumps(value, indent=2, ensure_ascii=False) + '\n', encoding='utf-8') def event(event_id, kind, payload): nonlocal state item = dict(schema=EVENT_SCHEMA, event_id=event_id, event_kind=kind, owner_surface=state['owner_surface'], previous_state_digest=state['state_digest'], payload=payload) save(event_id + '.event.json', item) state = apply_event(state, item) save(event_id + '.state.json', state) def source(event_id, text): event(event_id, 'source_event', dict(provenance_class='direct_operator', source_ref='inline:example', exact_text=text, source_digest=hashlib.sha256(text.encode('utf-8')).hexdigest(), secret_classified=False, source_gap_marker='', operative_status='active')) save('initial.state.json', state) source('S1', 'Prepare a draft for review. Do not publish it.') event('P1', 'assistant_projection', dict(projection='Prepare a public release now.', derives_from=['S1'], operative_status='active')) event('M1', 'verified_movement', dict(action_requested='prepare draft', action_performed='saved draft', evidence_refs=['illustrative:review-draft.md'], effects_observed=['draft-saved'], authority_before='project owner', authority_after='project owner')) source('S2', 'Correct the interpretation: the deliverable remains a draft for review.') event('R1', 'retraction_or_supersession', dict(retraction_authority_ref='S2', retracts=['P1'], scope='assistant interpretation', effective_relation_change='review draft, without publication', effects_not_reversed=['draft-saved'])) event('C1', 'relation_checkpoint', dict(source_event_refs=['S1', 'S2'], projection_refs=[], current_relation='Draft saved; owner review remains the selected next movement.', unknowns=[], next_legal_action='read the saved draft with its owner', reopen_condition='new owner instruction or source change')) receipt = reconstruct(state) save('reconstruction.json', receipt) print(json.dumps({'directory': str(output), 'events': len(state['events']), 'state': receipt['finite_state'], 'effects_preserved': receipt['effects_preserved']}, indent=2)) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--output-dir', type=Path, required=True, help='New directory; existing directories are not overwritten.') run(parser.parse_args().output_dir) ``` ## references/material-work-continuity.v0.schema.json ```json { "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "urn:maios:material-work-continuity:v1", "title": "D-ND Material Work Continuity", "type": "object", "additionalProperties": false, "required": [ "schema", "version", "owner_surface", "owner_state_ref", "authority_initial", "authority_current", "events", "active_event_ids", "superseded_event_ids", "unknowns", "current_relation", "next_legal_action", "reopen_condition", "finite_state", "previous_digest", "state_digest" ], "properties": { "schema": { "const": "dnd.material_work_continuity.state.v1" }, "version": { "const": 1 }, "owner_surface": { "type": "string", "minLength": 1 }, "owner_state_ref": { "type": "string", "minLength": 1 }, "authority_initial": { "type": "string", "minLength": 1 }, "authority_current": { "type": "string", "minLength": 1 }, "events": { "type": "array", "items": { "$ref": "#/$defs/event" } }, "active_event_ids": { "type": "array", "items": { "type": "string" }, "uniqueItems": true }, "superseded_event_ids": { "type": "array", "items": { "type": "string" }, "uniqueItems": true }, "unknowns": { "type": "array", "items": { "type": "string" }, "uniqueItems": true }, "current_relation": { "type": "string" }, "next_legal_action": { "type": "string" }, "reopen_condition": { "type": "string" }, "finite_state": { "enum": [ "EMPTY_CONTINUITY", "CONTINUITY_CAPTURED", "CURRENT_RELATION_RECONSTRUCTABLE", "INCOMPLETE_SOURCE_RECONSTRUCTION" ] }, "previous_digest": { "oneOf": [ { "type": "null" }, { "type": "string", "pattern": "^[0-9a-f]{64}$" } ] }, "state_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" } }, "$defs": { "stringList": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true }, "sourcePayload": { "type": "object", "additionalProperties": false, "required": [ "provenance_class", "source_ref", "exact_text", "source_digest", "secret_classified", "source_gap_marker", "operative_status" ], "properties": { "provenance_class": { "enum": [ "direct_operator", "owner_source", "operator_submitted_uncertain_authorship" ] }, "source_ref": { "type": "string", "minLength": 1 }, "exact_text": { "type": "string" }, "source_digest": { "type": "string" }, "secret_classified": { "type": "boolean" }, "source_gap_marker": { "type": "string" }, "operative_status": { "const": "active" } }, "allOf": [ { "if": { "properties": { "secret_classified": { "const": true } }, "required": [ "secret_classified" ] }, "then": { "properties": { "exact_text": { "const": "" }, "source_digest": { "const": "" }, "source_gap_marker": { "type": "string", "minLength": 1 } } }, "else": { "properties": { "source_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" }, "source_gap_marker": { "const": "" } } } }, { "if": { "properties": { "secret_classified": { "const": false }, "exact_text": { "const": "" } }, "required": [ "secret_classified", "exact_text" ] }, "then": { "properties": { "source_ref": { "type": "string", "pattern": "^(?:/|[A-Za-z]:[/\\\\])" } } } } ] }, "projectionPayload": { "type": "object", "additionalProperties": false, "required": [ "projection", "derives_from", "operative_status" ], "properties": { "projection": { "type": "string", "minLength": 1 }, "derives_from": { "$ref": "#/$defs/stringList", "minItems": 1 }, "operative_status": { "const": "active" } } }, "movementPayload": { "type": "object", "additionalProperties": false, "required": [ "action_requested", "action_performed", "evidence_refs", "effects_observed", "authority_before", "authority_after" ], "properties": { "action_requested": { "type": "string", "minLength": 1 }, "action_performed": { "type": "string", "minLength": 1 }, "evidence_refs": { "$ref": "#/$defs/stringList", "minItems": 1 }, "effects_observed": { "$ref": "#/$defs/stringList" }, "authority_before": { "type": "string", "minLength": 1 }, "authority_after": { "type": "string", "minLength": 1 } } }, "retractionPayload": { "type": "object", "additionalProperties": false, "required": [ "retraction_authority_ref", "retracts", "scope", "effective_relation_change", "effects_not_reversed" ], "properties": { "retraction_authority_ref": { "type": "string", "minLength": 1 }, "retracts": { "$ref": "#/$defs/stringList", "minItems": 1 }, "scope": { "type": "string", "minLength": 1 }, "effective_relation_change": { "type": "string", "minLength": 1 }, "effects_not_reversed": { "$ref": "#/$defs/stringList" } } }, "authorityPayload": { "type": "object", "additionalProperties": false, "required": [ "authority_before", "authority_after", "owner_receipt_ref" ], "properties": { "authority_before": { "type": "string", "minLength": 1 }, "authority_after": { "type": "string", "minLength": 1 }, "owner_receipt_ref": { "type": "string", "minLength": 1 } } }, "unknownPayload": { "type": "object", "additionalProperties": false, "required": [ "adds", "resolves", "evidence_refs" ], "properties": { "adds": { "$ref": "#/$defs/stringList" }, "resolves": { "$ref": "#/$defs/stringList" }, "evidence_refs": { "$ref": "#/$defs/stringList" } }, "allOf": [ { "if": { "properties": { "resolves": { "minItems": 1 } }, "required": [ "resolves" ] }, "then": { "properties": { "evidence_refs": { "minItems": 1 } } } } ] }, "checkpointPayload": { "type": "object", "additionalProperties": false, "required": [ "source_event_refs", "projection_refs", "current_relation", "unknowns", "next_legal_action", "reopen_condition" ], "properties": { "source_event_refs": { "$ref": "#/$defs/stringList", "minItems": 1 }, "projection_refs": { "$ref": "#/$defs/stringList" }, "current_relation": { "type": "string", "minLength": 1 }, "unknowns": { "$ref": "#/$defs/stringList" }, "next_legal_action": { "type": "string", "minLength": 1 }, "reopen_condition": { "type": "string", "minLength": 1 } } }, "event": { "type": "object", "additionalProperties": false, "required": [ "schema", "event_id", "event_kind", "owner_surface", "previous_state_digest", "payload" ], "properties": { "schema": { "const": "dnd.material_work_continuity.event.v1" }, "event_id": { "type": "string", "minLength": 1 }, "event_kind": { "enum": [ "source_event", "assistant_projection", "verified_movement", "retraction_or_supersession", "authority_change", "unknown_change", "relation_checkpoint" ] }, "owner_surface": { "type": "string", "minLength": 1 }, "previous_state_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" }, "payload": { "type": "object" } }, "allOf": [ { "if": { "properties": { "event_kind": { "const": "source_event" } } }, "then": { "properties": { "payload": { "$ref": "#/$defs/sourcePayload" } } } }, { "if": { "properties": { "event_kind": { "const": "assistant_projection" } } }, "then": { "properties": { "payload": { "$ref": "#/$defs/projectionPayload" } } } }, { "if": { "properties": { "event_kind": { "const": "verified_movement" } } }, "then": { "properties": { "payload": { "$ref": "#/$defs/movementPayload" } } } }, { "if": { "properties": { "event_kind": { "const": "retraction_or_supersession" } } }, "then": { "properties": { "payload": { "$ref": "#/$defs/retractionPayload" } } } }, { "if": { "properties": { "event_kind": { "const": "authority_change" } } }, "then": { "properties": { "payload": { "$ref": "#/$defs/authorityPayload" } } } }, { "if": { "properties": { "event_kind": { "const": "unknown_change" } } }, "then": { "properties": { "payload": { "$ref": "#/$defs/unknownPayload" } } } }, { "if": { "properties": { "event_kind": { "const": "relation_checkpoint" } } }, "then": { "properties": { "payload": { "$ref": "#/$defs/checkpointPayload" } } } } ] } } } ``` # Your context Intention, sources, existing work, available means and entrusted actions: [add here].