Applica questa competenza al lavoro descritto qui sotto. L'edizione pubblica 0.3.2 include metodo, sapere operativo ed esempi articolati. Aggiungi il tuo contesto dopo il sapere fornito. Usa strumenti esecutivi solo se questo host li fornisce.

# D-ND Material Work Continuity

**Ricostruire il lavoro e le decisioni di un progetto AI**

Conserva richieste originali, azioni, correzioni e punto attuale senza riscrivere ciò che è avvenuto.

## Scopo e situazione

Mantiene una traccia degli eventi materiali e distingue la fonte dell’operatore dalle interpretazioni dell’assistente e dagli effetti osservati.

Usa questa competenza: quando cambiano requisiti e direzione o serve ricostruire perché un progetto si trova nello stato attuale.

Una decisione può cambiare senza cancellare ciò che è avvenuto prima. Questa competenza conserva il collegamento fra richiesta originale, interpretazione dell'AI, lavoro svolto e correzione successiva.

Il risultato è una ricostruzione utilizzabile da chi riprende il progetto. La storia dettagliata resta raggiungibile; il punto corrente contiene quanto serve per continuare.

## Conservare la fonte prima del racconto

Quando una richiesta, una decisione o una correzione cambia materialmente il lavoro, conserva le parole originali o un riferimento esatto alla fonte accessibile nel progetto. Mantieni incertezze e formulazioni insolite. Attribuisci separatamente il riassunto dell'assistente e le sue inferenze. Se la fonte manca, dichiara la parte non ricostruibile: non ricreare una citazione da un ricordo o da un riepilogo. Usa il luogo di continuità già presente quando è adeguato.

## Seguire eventi ed effetti

Registra soltanto gli eventi che cambiano la ricostruzione: una fonte, una proposta dell'AI, un'azione con riscontro, una correzione, un cambiamento delle autorizzazioni o un'incognita decisiva. Collega ogni effetto alla sua prova e al momento in cui è stato osservato. Un file preparato, un invio eseguito e una risposta ricevuta sono fatti diversi. Conserva l'azione effettiva anche se il piano che l'ha prodotta viene successivamente abbandonato.

## Correggere senza riscrivere il passato

Una correzione aggiunge una nuova relazione alla storia: indica quale interpretazione o istruzione supera e quale seguito cambia. Non cancellare l'evento precedente né attribuire al passato una comprensione emersa dopo. La revoca di un permesso cambia ciò che si può fare ora, non annulla un effetto già avvenuto. Una sintesi dell'AI non può creare un'autorizzazione: conserva il riferimento alla decisione della persona o del responsabile effettivo.

## Rendere ricostruibile il punto corrente

Mantieni un ingresso breve che colleghi fonti ancora valide, decisioni, effetti compiuti, incognite e prossimo movimento pertinente. Il registro dettagliato e l'ingresso corrente svolgono funzioni diverse. Se la complessità lo richiede, usa eventi identificati e riferimenti ai predecessori; altrimenti bastano note attribuite e datate. Nel riprendere verifica che i riferimenti utili siano raggiungibili e segnala contenuti mancanti o cambiati. Non dichiarare copertura di una fonte non letta.

## Proteggere il lavoro e migliorare la ricostruzione

Scrivi nel progetto selezionato e preserva contributi concorrenti; se la destinazione di persistenza manca, prepara una nota che la persona possa conservare. Non costruire un diario universale. Escludi segreti e dati privati estranei, senza conservarne impronte derivate. Quando una ripresa rivela una perdita causale, correggi il modo in cui la continuità viene mantenuta: la nuova nota deve permettere una ricostruzione migliore, non soltanto aggiungere volume.

## Profondità del metodo

Questa competenza opera sulla continuità materiale: conserva che cosa è accaduto e da quali fonti deriva. Non decide da sola il significato di una scelta né lo stato complessivo del progetto.

La relazione diventa meta quando un errore di ricostruzione cambia il metodo con cui si conserva il lavoro. Per esempio, un riepilogo aveva mescolato proposta e approvazione: il seguito rende riconoscibili entrambe. Non è autonomia esecutiva, e un registro ben formato non dimostra memoria infallibile.

## Continuità e apprendimento

Un passaggio di consegne mostra quali collegamenti sono sufficienti e quali mancano. Conserva la correzione nel metodo di continuità, lasciando le scelte di progetto al loro responsabile.

Nel primo uso, parti dal contesto e dalle fonti effettivamente disponibili. Il metodo può essere esercitato in chat; persistenza, strumenti e azioni esterne dipendono dai mezzi e dal mandato del destinatario. Le istruzioni rendono la capacità utilizzabile, mentre il suo comportamento e il suo valore richiedono osservazione nel lavoro reale.
## Sapere operativo raggiungibile

Raggiungi [GUIDA_OPERATIVA_IT.md](GUIDA_OPERATIVA_IT.md) come parte di questo metodo mentre formi il lavoro. Conserva sapere pratico, relazioni decisionali e mezzi della capacità. L'avvio completo include questo sapere e i casi articolati, rendendolo utilizzabile senza fonti private. Conserva l'identità sorgente pubblica e adatta attraverso i mezzi reali del ricevente.

# D-ND Material Work Continuity

## Usare il registratore deterministico quando conta l'integrità esatta degli eventi

Il metodo testuale si usa anche da solo. Il kit fornisce inoltre reducer Python, schema JSON di eventi/stato, un registro d'esempio e test portabili della competenza sorgente. Il reducer verifica l'integrità del registro, senza diventare un'ammissione semantica o un'autorità indipendente. Gli hash rilevano registrazioni incoerenti; non provano la verità di una frase o di un riferimento di evidenza. Conserva registro originale e riferimenti alle fonti nel progetto proprietario.

I sette tipi di evento conservano fonte esatta, proiezione dell'assistente, movimento osservato, ritrattazione o superamento, cambio di autorità, cambiamento delle incognite e punto della relazione corrente. Le fonti mantengono l'attribuzione. Le proiezioni citano eventi precedenti. Una correzione supera un'interpretazione senza cancellare l'evento o gli effetti osservati. Il checkpoint espone fonti correnti, incognite residue e prossima azione permessa. Le fonti classificate segrete conservano un indicatore di lacuna, senza valori o hash del segreto. Non copiare credenziali nel registro.

## Eseguire l'esempio incluso

Usa Python 3.10 o successivo dalla cartella della competenza. Esegui `python scripts/demo_continuity.py --output-dir demo-ledger` scegliendo una cartella nuova. Scrive stato iniziale, sei eventi illustrativi con i rispettivi stati successivi e `reconstruction.json`. L'esempio corregge l'interpretazione di pubblicazione dell'assistente conservando una bozza realmente salvata. Non pubblica né contatta nessuno.

Esegui `python scripts/reduce_material_work_continuity.py verify --state demo-ledger/C1.state.json` per ricostruire quello stato. Per il tuo progetto, `init` accetta `--owner-surface`, `--owner-state-ref`, `--authority` e `--output` facoltativo. `reduce` accetta `--state`, `--event` e `--output` facoltativo; usa un file di uscita separato per conservare lo stato precedente. `previous_state_digest` dell'evento deve corrispondere allo stato che estende. L'esempio mostra la costruzione degli eventi attraverso l'interfaccia Python.

## Comprendere errori e limiti

La ripetizione esatta è idempotente; un evento alterato con la stessa identità viene respinto. Sono respinti predecessore superato, stato alterato, riferimento di autorità fabbricato o risoluzione di un'incognita priva di evidenza. Una fonte strutturata di concessione dell'autorità modella il cambiamento registrato; lo strumento non autentica la persona dietro un documento esterno. La fonte referenziata deve essere un file regolare assoluto con il digest registrato, al massimo 1 MiB; riferimenti mancanti, simbolici o cambiati rendono incompleta la ricostruzione. Le fonti esatte inline restano nel registro.

La CLI scrive soltanto l'uscita scelta e non usa la rete. Scrittori concorrenti richiedono serializzazione nel reale file o archivio; il reducer da solo non fornisce lock o transazioni. Lo stato valido per schema viene controllato anche ripercorrendo la catena degli eventi. Esegui i test portabili con `python -m pip install -r requirements-test.txt` e poi `python scripts/test_reduce_material_work_continuity.py`. Il reducer usa solo la libreria standard; `jsonschema` serve ai test dello schema. Lo schema mantiene i nomi del protocollo originale per compatibilità, con identificatore pubblico e percorsi Windows/POSIX.

# Esempi — Ricostruire il lavoro e le decisioni di un progetto AI

I tre casi sono illustrativi: mostrano come il metodo cambia il lavoro. Non sono dichiarazioni di risultati verificati in installazioni esterne.

## 1. Una scelta del cliente cambia

Il cliente chiede una pagina unica e approva una bozza. In seguito sceglie due percorsi distinti. Conserva richiesta e approvazione originali, bozza prodotta e nuova decisione; indica quali lavori restano utilizzabili. Il punto corrente porta alla struttura nuova senza far apparire sbagliata o inesistente l'approvazione precedente.

Per provarlo, fornisci le fonti pertinenti del tuo caso e chiedi all'AI di applicare il metodo, conservando ciò che è osservato e ciò che resta da capire.

## 2. Un invio precede una correzione

Un documento è stato inviato prima che una cifra venga corretta. Registra l'identità della versione inviata e l'effetto osservato, poi la correzione. La versione locale aggiornata non sostituisce il documento già ricevuto. Un eventuale nuovo invio richiede il mandato pertinente.

Per provarlo, fornisci le fonti pertinenti del tuo caso e chiedi all'AI di applicare il metodo, conservando ciò che è osservato e ciò che resta da capire.

## 3. Un nuovo assistente riprende il progetto

La nuova chat riceve fonti, decisioni e azioni documentate. Una conversazione citata non è più accessibile: segnala il vuoto e continua dalle informazioni disponibili. La ricostruzione distingue ciò che può essere verificato da ciò che è conosciuto soltanto attraverso una sintesi.

Per provarlo, fornisci le fonti pertinenti del tuo caso e chiedi all'AI di applicare il metodo, conservando ciò che è osservato e ciò che resta da capire.

## Continuare dal proprio caso

Un passaggio di consegne mostra quali collegamenti sono sufficienti e quali mancano. Conserva la correzione nel metodo di continuità, lasciando le scelte di progetto al loro responsabile.

[Metodo completo](SKILL_IT.md) · [Avvio da usare in chat](AVVIO_CHAT_IT.txt)


# Sorgenti eseguibili incluse


## 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"
              }
            }
          }
        }
      ]
    }
  }
}
```


# Il tuo contesto

Intento, fonti, lavoro esistente, mezzi disponibili e azioni affidate: [aggiungi qui].
