notebook.py

"""# ADCS Lifecycle Demo: Bidirectional Requirements Traceability

A walkthrough of satellite attitude control system design — from receiving
requirements through symbolic analysis, numerical simulation, evidence
binding, human attestation, and audit.
"""

import marimo

__generated_with = "0.22.5"
app = marimo.App(width="medium")


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    # ADCS Lifecycle Demo

    ## Bidirectional Requirements Traceability with Reproducible Evidence

    This notebook walks through the complete lifecycle of verifying an
    **Attitude Determination and Control System (ADCS)** for a geostationary
    communications satellite.

    We follow the perspective of the **controls engineering team** — one
    disciplinary team within a larger satellite design program. Our job is
    to demonstrate that the ADCS meets its requirements, with evidence that
    any auditor can interrogate and reproduce.

    ### Core Principle

    > **Evidence does not verify requirements; evidence supports a human
    > judgment that requirements are satisfied.**

    Models are imperfect representations of physical systems. Symbolic proofs
    and simulation results are claims true *within the model*. The engineer
    judges model adequacy and evidence sufficiency. Only human attestation
    connects evidence to requirement satisfaction.
    """)
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Prologue: The Integration Ontology

    Before we write a single line of analysis code, we have an
    epistemological question to settle: **what counts as a satisfactory
    requirement?**

    A naive demo would invent terms — `rtm:modelAdequacy`, a custom
    "satisfaction" predicate, a bespoke evidence vocabulary. This demo
    deliberately does the opposite. The `rtm:` namespace introduces **no
    novel epistemic vocabulary.** It is a thin *integration ontology* over
    established standards:

    | Layer        | Vocabulary                           | Role                                                     |
    | ------------ | ------------------------------------ | -------------------------------------------------------- |
    | W3C / IETF   | `prov:`, `dcterms:`, `earl:`, `sh:`  | Provenance + assertion + outcome + SHACL closure         |
    | OMG / SysML  | `sysml:` ↔ `omg-sysml:`              | Structural model (aliased to OMG SysMLv2 OWL rendering)  |
    | Community    | `gsn:`, `p-plan:`                    | Assurance argument structure + declarative process model |
    | Tool interop | `oslc_rm:`, `oslc_qm:`               | Aliases for DOORS Next / Jama / RQM                      |

    The adequacy/sufficiency split is **not novel** either — it's the
    canonical Hawkins–Habli Assurance Claim Point categorization.
    "Adequacy" is a `gsn:Assumption`; "sufficiency" is a
    `gsn:Justification`. Both attach to the attestation via
    `gsn:inContextOf`. The text content lives on those GSN nodes in
    `gsn:statement`.

    The pipeline runs this assembly as its first act — narrating which
    upstream ontologies were imported, how many terms we reference, and
    what closure rules will be enforced downstream.
    """)
    return


@app.cell(hide_code=True)
def __(mo):
    import json as _json
    from pathlib import Path as _Path

    _manifest = _json.loads(_Path("ontology/assembly_manifest.json").read_text())

    _import_rows = []
    for _name in sorted(_manifest["imports"]):
        _info = _manifest["imports"][_name]
        _import_rows.append(
            f"| {_name} | {_info['total_triples']:>5} | {_info['referenced_count']} |"
        )

    mo.md(
        "### Assembly manifest (data-driven, not hand-written)\n\n"
        f"Built `{_manifest['build_time']}` from `ontology/rtm-edit.ttl`.\n\n"
        "| Upstream | Triples | TBox refs in `rtm-edit.ttl` |\n"
        "|---|---:|---:|\n"
        + "\n".join(_import_rows) + "\n\n"
        f"- **SysMLv2 equivalence axioms** (sysml: ↔ omg-sysml:): "
        f"{_manifest['artifact']['equivalence_axioms']}\n"
        f"- **Local rtm: integration glue:** "
        f"{_manifest['artifact']['subclass_axioms']} subclass + "
        f"{_manifest['artifact']['subproperty_axioms']} subproperty axioms "
        f"(no novel epistemic terms)\n"
        f"- **Artifact SHA-256:** `{_manifest['artifact']['sha256'][:24]}...`\n\n"
        "**About the third column.** *TBox refs* counts how many distinct "
        "terms from each upstream namespace appear in our integration "
        "ontology source `rtm-edit.ttl` — i.e. how often that vocabulary "
        "is used as the target of a `rdfs:subClassOf` / `rdfs:subPropertyOf` "
        "alignment axiom. **P-PLAN reads 0** because P-PLAN is used at the "
        "*instance / runtime* layer rather than the TBox alignment layer: "
        "the plan definition lives in `pipeline/plan.ttl` (instance data, "
        "10 `p-plan:Step` instances) and per-stage `p-plan:Activity` "
        "triples are emitted at runtime by `traceability.plan_execution`. "
        "Vendoring an upstream is not the same as subclassing it.\n\n"
        "The manifest is the build-step provenance record. Stage 0 of the "
        "pipeline verifies that `rtm.ttl` still hashes to this manifest "
        "value — drift between the source `rtm-edit.ttl` and the committed "
        "artifact fails the pipeline with a clear remediation hint."
    )
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ### Named-graph quadstore layout

    The runtime holds the RTM as an `rdflib.Dataset` (a quadstore) with
    one named graph per content layer — sized to match how Flexo MMS
    partitions projects/branches. SPARQL queries use
    `Dataset(default_union=True)` so existing queries match across the
    union without `GRAPH` clauses.

    ```text
    <rtm:ontology>        TBox + shapes + individuals
    <rtm:plan>            P-PLAN process model (one Step per pipeline stage)
    <adcs:structural>     SysMLv2 instance data
    <adcs:context>        Stable gsn:Context / gsn:Assumption individuals
    <adcs:evidence>       rtm:Evidence artifacts
    <adcs:attestations>   rtm:Attestation events
    <adcs:plan-execution> p-plan:Activity instances (one per stage)
    <adcs:audit>          Forward/backward/bidirectional audit summary
    ```

    Stage 7 persists the Dataset to disk (`output/rtm.{ttl,trig}`) or to a
    real quadstore (Flexo MMS / Apache Jena Fuseki) via pluggable
    backends. Either way, every named graph round-trips cleanly.
    """)
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 1: The Assignment

    You are Dr. Michael Zargham, lead controls engineer on the GeoSat
    communications satellite program. Systems engineering has allocated
    four requirements to your ADCS subsystem, each derived from
    satellite-level requirements.

    Your team owns the ADCS — reaction wheels, star tracker, IMU, and the
    PD attitude controller. You consume interface parameters (mass, orbit,
    panel geometry) from other teams but don't control them.

    ```
    Satellite (system-of-interest)
    ├── ADCS           ← YOUR SCOPE
    ├── Power          ← interface: power budget
    ├── Communications ← interface: antenna pointing
    ├── Thermal        ← interface: wheel heat dissipation
    └── Structure      ← interface: mass properties
    ```

    Let's load the structural model and see what we're working with.
    """)
    return


@app.cell(hide_code=True)
def __():
    import sys
    sys.path.insert(0, ".")

    from rdflib import Graph
    from ontology.prefixes import bind_prefixes, SYSML, RTM, ADCS, SAT, PROV
    from traceability.queries import query_to_dicts
    return Graph, bind_prefixes, SYSML, RTM, ADCS, SAT, PROV, query_to_dicts, sys


@app.cell(hide_code=True)
def __(Graph, bind_prefixes):
    from analysis.load_params import load_structural_graph, load_params

    struct_graph = load_structural_graph()
    params = load_params(struct_graph)
    return struct_graph, params, load_structural_graph, load_params


@app.cell(hide_code=True)
def __(mo, params):
    _param_rows = "\n".join(
        f"| {k} | {v:.6g} |" for k, v in sorted(params.items())
    )
    mo.md(
        "### Structural Parameters (from RDF via SPARQL)\n\n"
        "All parameters flow from the SysMLv2 structural model — nothing is "
        "hardcoded. If systems engineering updates the satellite mass, our "
        "entire analysis chain re-derives from the new value.\n\n"
        "| Parameter | Value |\n"
        "|-----------|-------|\n"
        f"{_param_rows}"
    )
    return


@app.cell(hide_code=True)
def __(mo, struct_graph, query_to_dicts):
    _req_query = """
    SELECT ?name ?text WHERE {
        ?req a sysml:RequirementDefinition ;
             sysml:declaredName ?name ;
             sysml:text ?text .
        FILTER(STRSTARTS(?name, "REQ-"))
    }
    ORDER BY ?name
    """
    _reqs = query_to_dicts(struct_graph, _req_query)

    _deriv_query = """
    SELECT ?child ?parent WHERE {
        ?c sysml:declaredName ?child ;
           rtm:derivedFrom ?p .
        ?p sysml:declaredName ?parent .
    }
    ORDER BY ?child
    """
    _derivs = query_to_dicts(struct_graph, _deriv_query)
    _deriv_map = {r["child"]: r["parent"] for r in _derivs}

    _alloc_query = """
    SELECT ?reqName ?elementName WHERE {
        ?req sysml:declaredName ?reqName ;
             sysml:ownedRelationship ?rel .
        ?rel a sysml:SatisfyRequirementUsage ;
             sysml:satisfyingElement ?el .
        ?el sysml:declaredName ?elementName .
        FILTER(STRSTARTS(?reqName, "REQ-"))
    }
    ORDER BY ?reqName ?elementName
    """
    _allocs = query_to_dicts(struct_graph, _alloc_query)

    _alloc_map = {}
    for _a in _allocs:
        _alloc_map.setdefault(_a["reqName"], []).append(_a["elementName"])

    _req_rows = []
    for _r in _reqs:
        _name = _r["name"]
        _text = _r["text"].strip().replace("\n", " ")[:80]
        _parent = _deriv_map.get(_name, "—")
        _elements = ", ".join(_alloc_map.get(_name, []))
        _req_rows.append(f"| {_name} | {_text}... | {_parent} | {_elements} |")

    _req_table = "\n".join(_req_rows)

    mo.md(
        "### ADCS Requirements\n\n"
        "Four requirements allocated to us, each derived from a satellite-level "
        "parent requirement and satisfied by specific design elements:\n\n"
        "| ID | Requirement | Derived From | Satisfied By |\n"
        "|----|------------|-------------|-------------|\n"
        f"{_req_table}"
    )
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 2: Symbolic Analysis

    Before running any simulation, we derive formal results symbolically
    using SymPy. Every quantity is computed from the structural parameters —
    the inertia tensor via parallel axis theorem, eigenvalues for stability
    analysis, and bounds for pointing error and wheel momentum.

    These are claims true *within our model*. Whether the model adequately
    represents the physical satellite is a judgment we'll make during
    attestation.
    """)
    return


@app.cell(hide_code=True)
def __(params):
    from analysis.symbolic import (
        run_symbolic_analysis,
        build_inertia_tensor_symbolic,
        evaluate_inertia,
        stability_margins,
    )

    sym_result = run_symbolic_analysis(params)
    return sym_result, run_symbolic_analysis, build_inertia_tensor_symbolic, evaluate_inertia, stability_margins


@app.cell(hide_code=True)
def __(mo, sym_result):
    _Ixx, _Iyy, _Izz = sym_result.inertia
    _margins = sym_result.stability_margins

    mo.md(f"""
    ### Composite Inertia Tensor

    Derived via parallel axis theorem (bus + 2 solar panels + antenna):

    | Axis | Inertia (kg-m^2) | Dominant contributor |
    |------|-----------------|---------------------|
    | Ixx  | {_Ixx:.1f} | Solar panels (offset along Y) |
    | Iyy  | {_Iyy:.1f} | Bus (panels add little on this axis) |
    | Izz  | {_Izz:.1f} | Solar panels (offset along Y) |

    The panels dominate Ixx and Izz because their center of mass is far
    from the satellite center — the parallel axis term grows as distance
    squared.

    ### Stability Margins (REQ-003)

    Closed-loop eigenvalues for each axis (PD controller, linearized):

    | Axis | Re(lambda) | Margin vs -0.010 |
    |------|-----------|-----------------|
    | X | {_margins['x']:.4f} rad/s | {'PASS' if _margins['x'] <= -0.010 else 'MARGINAL'} |
    | Y | {_margins['y']:.4f} rad/s | {'PASS' if _margins['y'] <= -0.010 else 'MARGINAL'} |
    | Z | {_margins['z']:.4f} rad/s | {'PASS' if _margins['z'] <= -0.010 else 'MARGINAL'} |

    All axes satisfy REQ-003 (Re(lambda) <= -0.010 rad/s).
    """)
    return


@app.cell(hide_code=True)
def __(mo, sym_result):
    _pb = sym_result.pointing_budget
    _gg = sym_result.gravity_gradient
    _wm = sym_result.wheel_momentum

    mo.md(f"""
    ### Pointing Budget (REQ-001)

    | Metric | Value |
    |--------|-------|
    | Steady-state error (gravity gradient) | {_pb['theta_ss_deg']:.6f} deg |
    | Star tracker noise floor | {_pb['st_floor_deg']:.6f} deg |
    | Settling time (4/|Re(lambda)|) | {_pb['settling_time_s']:.1f} s |

    The steady-state pointing error is well below 0.1 deg. However, the
    **settling time is {_pb['settling_time_s']:.0f}s** — exceeding the 120s target.
    This is a real finding that the engineer must address during attestation.

    ### Gravity Gradient (REQ-004)

    | Metric | Value |
    |--------|-------|
    | tau_gg_x | {_gg['tau_gg_x']:.2e} N.m |
    | tau_gg_y | {_gg['tau_gg_y']:.2e} N.m |
    | Actuator capacity | {_gg['tau_max']} N.m |

    Gravity gradient torques at GEO are **orders of magnitude** below
    actuator capacity.

    ### Wheel Momentum (REQ-002)

    | Metric | Value |
    |--------|-------|
    | Peak momentum (10 deg slew) | {_wm['h_peak']:.3f} N.m.s |
    | Rated capacity | {_wm['h_max']} N.m.s |
    | Margin | {_wm['margin']:.3f} N.m.s |
    """)
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ### Formal Proofs

    Each requirement gets a ProofScript — a chain of SymPy lemmas, each
    independently re-verifiable. The proof is bound to the structural model
    via content hash: if the model changes, the proof hash changes, alerting
    auditors to re-verify.
    """)
    return


@app.cell(hide_code=True)
def __(struct_graph):
    from evidence.hashing import hash_structural_model, hash_proof
    from analysis.build_proofs import build_all_proofs
    from analysis.proof_scripts import verify_proof, ProofStatus

    model_hash = hash_structural_model(struct_graph)
    proofs = build_all_proofs(model_hash)

    proof_results = {}
    for _req_id, _script in proofs.items():
        _result = verify_proof(_script, model_hash)
        proof_results[_req_id] = _result

    return model_hash, proofs, proof_results, hash_structural_model, hash_proof, build_all_proofs, verify_proof, ProofStatus


@app.cell(hide_code=True)
def __(mo, proofs, proof_results, ProofStatus):
    _rows = []
    for _req_id in sorted(proofs.keys()):
        _script = proofs[_req_id]
        _result = proof_results[_req_id]
        _status = "VERIFIED" if _result.status == ProofStatus.VERIFIED else "FAILED"
        _lemmas = ", ".join(l.name for l in _script.lemmas)
        _rows.append(f"| {_req_id} | {_status} | {_script.claim[:60]}... | {_lemmas} |")

    _proof_table = "\n".join(_rows)
    mo.md(
        "| Requirement | Status | Claim | Lemmas |\n"
        "|-------------|--------|-------|--------|\n"
        f"{_proof_table}\n\n"
        "All proofs pass. Each can be serialized to JSON, stored, and re-verified "
        "by anyone — no trust in the original analyst required."
    )
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 3: Numerical Simulation

    Symbolic analysis tells us what the model *should* do. Numerical
    simulation shows what it *actually does* when we integrate the full
    nonlinear dynamics. We run two scenarios:

    1. **Step response** — 10-degree initial attitude error, observe settling
    2. **Disturbance rejection** — near-zero error, observe gravity gradient effects
    """)
    return


@app.cell(hide_code=True)
def __(params):
    from analysis.numerical import run_step_response, run_disturbance_rejection

    step_result = run_step_response(params)
    step_summary = step_result.summary()

    dist_result = run_disturbance_rejection(params)
    dist_summary = dist_result.summary()
    return step_result, step_summary, dist_result, dist_summary, run_step_response, run_disturbance_rejection


@app.cell(hide_code=True)
def __(mo, step_result, step_summary):
    import matplotlib.pyplot as plt
    import numpy as np

    _fig, _axes = plt.subplots(2, 2, figsize=(12, 8))
    _axis_colors = {"X": "#1f77b4", "Y": "#2ca02c", "Z": "#9467bd"}  # blue, green, purple
    _limit_color = "#d62728"  # red reserved exclusively for requirement limits

    # Attitude error
    _q_vec = np.linalg.norm(step_result.q[:, :3], axis=1)
    _theta_deg = np.degrees(2 * _q_vec)
    _axes[0, 0].semilogy(step_result.t, _theta_deg, color=_axis_colors["X"], linewidth=1.5, label='Attitude error')
    _axes[0, 0].axhline(0.1, color=_limit_color, linestyle='--', linewidth=1, label='REQ-001 limit (0.1 deg)')
    _axes[0, 0].set_xlabel('Time (s)')
    _axes[0, 0].set_ylabel('Attitude Error (deg)')
    _axes[0, 0].set_title('Pointing Convergence')
    _axes[0, 0].set_ylim(bottom=1e-3)
    _axes[0, 0].legend(fontsize=8)
    _axes[0, 0].grid(True, alpha=0.3, which='both')

    # Angular velocity
    for _i, (_axis, _c) in enumerate(_axis_colors.items()):
        _axes[0, 1].plot(step_result.t, np.degrees(step_result.omega[:, _i]),
                         color=_c, linewidth=1, label=f'{_axis}-axis')
    _axes[0, 1].set_xlabel('Time (s)')
    _axes[0, 1].set_ylabel('Angular Rate (deg/s)')
    _axes[0, 1].set_title('Angular Velocity')
    _axes[0, 1].legend(fontsize=8)
    _axes[0, 1].grid(True, alpha=0.3)

    # Control torque
    for _i, (_axis, _c) in enumerate(_axis_colors.items()):
        _axes[1, 0].plot(step_result.t, step_result.tau_ctrl[:, _i],
                         color=_c, linewidth=1, label=f'{_axis}-axis')
    _axes[1, 0].axhline(step_result.config.max_torque, color=_limit_color, linestyle='--',
                         linewidth=1, alpha=0.7, label='Torque limit')
    _axes[1, 0].axhline(-step_result.config.max_torque, color=_limit_color, linestyle='--',
                         linewidth=1, alpha=0.7)
    _axes[1, 0].set_xlabel('Time (s)')
    _axes[1, 0].set_ylabel('Torque (N.m)')
    _axes[1, 0].set_title('Control Torque')
    _axes[1, 0].legend(fontsize=8)
    _axes[1, 0].grid(True, alpha=0.3)

    # Wheel momentum
    _h_mag = np.linalg.norm(step_result.h_wheel, axis=1)
    _axes[1, 1].plot(step_result.t, _h_mag, color=_axis_colors["X"], linewidth=1.5, label='|h| (total)')
    _axes[1, 1].axhline(step_result.config.max_momentum, color=_limit_color, linestyle='--',
                         linewidth=1, label='REQ-002 limit (4.0 N.m.s)')
    _axes[1, 1].set_xlabel('Time (s)')
    _axes[1, 1].set_ylabel('Momentum (N.m.s)')
    _axes[1, 1].set_title('Wheel Angular Momentum')
    _axes[1, 1].legend(fontsize=8)
    _axes[1, 1].grid(True, alpha=0.3)

    _fig.suptitle('Step Response: 10-degree Slew Maneuver', fontsize=14, fontweight='bold')
    plt.tight_layout()

    mo.md(f"""
    ### Step Response Results

    | Metric | Value |
    |--------|-------|
    | Final pointing error | {step_summary['final_error_deg']:.4f} deg |
    | Peak pointing error | {step_summary['peak_error_deg']:.1f} deg |
    | Settling time | {step_summary['settling_time_s']:.1f} s |
    | Peak wheel momentum | {step_summary['peak_wheel_momentum']:.3f} N.m.s |
    | Peak control torque | {step_summary['peak_control_torque']:.4f} N.m |
    """)

    _fig
    return np, plt


@app.cell(hide_code=True)
def __(mo, dist_summary):
    mo.md(f"""
    ### Disturbance Rejection Results

    | Metric | Value |
    |--------|-------|
    | Peak error (GG disturbance) | {dist_summary['peak_error_deg']:.6f} deg |
    | Final angular rate | {dist_summary['final_omega_norm']:.2e} rad/s |

    Gravity gradient effects are negligible at GEO — confirming REQ-004.
    """)
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 4: Evidence Binding

    Now we bind our computational results to the RDF traceability graph.
    Every evidence artifact gets a content hash, a model hash (binding it
    to the structural model version), and PROV-O provenance (who/what
    produced it, when).

    Each evidence artifact **addresses** a specific requirement — recording
    the structural intent that "this proof was constructed to evaluate
    REQ-003." But `rtm:addresses` is not `rtm:attests`. The evidence
    says *what was analyzed*; only human attestation says *whether it's
    sufficient*. An evidence artifact can address a requirement and still
    lead to a declined attestation — as we'll see with REQ-001.
    """)
    return


@app.cell(hide_code=True)
def __(mo, model_hash, proofs, proof_results, step_summary, dist_summary, params, ProofStatus):
    from evidence.binding import bind_proof_evidence, bind_simulation_evidence, bind_computation_engines
    from evidence.hashing import hash_proof as _hp, hash_evidence, hash_simulation
    from pipeline.dataset import graph_for, triples_by_graph
    from traceability.rtm import load_base_dataset, verify_evidence_completeness

    # rtm_graph is now an rdflib.Dataset with named graphs. Existing
    # SPARQL queries still work via default_union; new audit / closure-
    # rule / backend code uses the explicit named-graph views.
    rtm_graph = load_base_dataset()
    _ev = graph_for(rtm_graph, "evidence")
    bind_computation_engines(_ev)

    for _rid, _script in proofs.items():
        _ph = _hp(_script, model_hash)
        _ch = hash_evidence(model_hash, proof_hash=_ph)
        bind_proof_evidence(
            _ev, f"EV-PROOF-{_rid}", f"SA-{_rid}", _rid,
            model_hash, _ph, _ch,
            f"Symbolic proof: {_script.claim}",
            source_file="analysis/build_proofs.py",
        )

    _sh = hash_simulation({"type": "step_response"}, step_summary)
    for _rid, _desc in [
        ("REQ-001", f"Step response: settling={step_summary['settling_time_s']:.1f}s, final_error={step_summary['final_error_deg']:.4f} deg"),
        ("REQ-002", f"Peak wheel momentum: {step_summary['peak_wheel_momentum']:.3f} N.m.s (limit={params['maxMomentum']})"),
    ]:
        bind_simulation_evidence(
            _ev, f"EV-SIM-{_rid}", f"NS-{_rid}", _rid,
            model_hash, _sh, _desc, source_file="analysis/numerical.py",
        )

    _dh = hash_simulation({"type": "disturbance_rejection"}, dist_summary)
    bind_simulation_evidence(
        _ev, "EV-SIM-REQ-004", "NS-REQ-004", "REQ-004",
        model_hash, _dh,
        f"Disturbance rejection: peak_error={dist_summary['peak_error_deg']:.6f} deg",
        source_file="analysis/numerical.py",
    )

    _issues = verify_evidence_completeness(rtm_graph)
    _counts = triples_by_graph(rtm_graph)

    _count_rows = "\n".join(
        f"| `<{_iri.rsplit('/', 1)[-1]}>` | {_n} |"
        for _iri, _n in sorted(_counts.items())
    )

    mo.md(
        "### Evidence artifacts in their named graph\n\n"
        f"- **4 proof artifacts** (hash-bound to model `{model_hash[:16]}...`)\n"
        f"- **3 simulation results**\n"
        f"- All emitted into `<adcs:evidence>` — kept distinct from the structural "
        f"and ontology layers so SPARQL queries can scope by graph, and the "
        f"Phase J Flexo backend can push each layer as its own branch.\n"
        f"- Evidence completeness: **{'PASS' if not _issues else 'ISSUES: ' + str(_issues)}**\n\n"
        "**Per-graph triple counts after Act 4:**\n\n"
        "| Named graph | Triples |\n"
        "|---|---:|\n"
        + _count_rows + "\n\n"
        "Every artifact carries `rtm:contentHash`, `rtm:modelHash`, and a "
        "PROV-O provenance chain. The model hash ensures that if the model "
        "changes (Act 8), all evidence must be re-produced and re-verified."
    )
    return rtm_graph, bind_proof_evidence, bind_simulation_evidence, bind_computation_engines, hash_evidence, hash_simulation, load_base_dataset, verify_evidence_completeness


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 5: Attestation (GSN + EARL outcomes)

    Evidence alone doesn't satisfy requirements. The engineer makes two
    judgments per requirement — and we record them using **established
    assurance-case vocabulary**, not novel terms:

    - **Adequacy** → a `gsn:Assumption` (Hawkins–Habli "asserted context")
      stating the model adequately represents the physical system for
      this requirement. Text on `gsn:statement`.
    - **Sufficiency** → a `gsn:Justification` (Hawkins–Habli "asserted
      inference") stating the evidence is sufficient to conclude
      satisfaction. Text on `gsn:statement`.
    - **Outcome** → an `earl:outcome` from EARL's five-valued lattice:
      `earl:passed` / `earl:failed` / `earl:cantTell` / `earl:inapplicable`
      / `earl:untested`. Better than binary pass/fail because "models are
      imperfect" — `cantTell` and `inapplicable` are first-class.
    - **Qualified association** → a `prov:Association` carrying the
      engineer's `prov:hadRole` (`rtm:role-AttestingEngineer`) and the
      `prov:hadPlan` they followed (the standard attestation procedure).

    REQ-001 below is **attested-with-failed**, not silently omitted —
    the audit trail records the declination as a well-formed attestation
    so closure-rule shapes can validate against an audit-complete graph.
    """)
    return


@app.cell(hide_code=True)
def __(rtm_graph, step_summary, params, mo):
    from traceability.attestation import request_attestation, OUTCOME_FAILED

    _adequacy = {
        "REQ-001": ("Step-response simulation is adequate for evaluating pointing-"
                    "accuracy settling time at this point in the lifecycle."),
        "REQ-002": ("Energy-based momentum bound is conservative. "
                    "Reaction wheel model adequate for peak momentum estimation."),
        "REQ-003": ("Linearized stability analysis via Routh-Hurwitz is adequate for this design point. "
                    "Nonlinear effects are second-order for small angles around the operating point."),
        "REQ-004": ("Linearized gravity gradient model adequate for GEO orbit. "
                    "Higher-order terms negligible at geostationary altitude."),
    }
    _sufficiency = {
        "REQ-001": (f"Evidence is sufficient to conclude REQ-001 is NOT yet satisfied: "
                    f"settling time {step_summary['settling_time_s']:.0f}s exceeds the 120s "
                    f"requirement. Action item: retune gains (Kp: {params['Kp']:.0f}→4, "
                    f"Kd: {params['Kd']:.0f}→30) and re-verify."),
        "REQ-002": ("Both symbolic bound (0.81 N.m.s) and numerical simulation confirm "
                    "peak momentum well below 4.0 N.m.s rated capacity. Large margin."),
        "REQ-003": ("Routh-Hurwitz proof confirms asymptotic stability for ALL positive J, Kp, Kd — "
                    "this is a parametric result, not just for one design point. "
                    "Numerical eigenvalues confirm margins exceed -0.010 rad/s on all axes."),
        "REQ-004": ("Gravity gradient torques at GEO are ~1e-6 N.m, four orders of magnitude below "
                    "0.1 N.m actuator capacity. Simulation confirms negligible pointing impact. "
                    "Overwhelming margin."),
    }

    # REQ-001: explicit DECLINATION as earl:failed — keeps the audit
    # trail complete so the closure-rule suite validates.
    request_attestation(
        rtm_graph, "REQ-001", "Dr. Michael Zargham (@mzargham)",
        auto_attest=True,
        model_adequacy=_adequacy["REQ-001"],
        evidence_sufficiency=_sufficiency["REQ-001"],
        outcome=OUTCOME_FAILED,
    )

    # REQ-002, REQ-003, REQ-004 — outcome defaults to earl:passed
    for _rid in ["REQ-002", "REQ-003", "REQ-004"]:
        request_attestation(
            rtm_graph, _rid, "Dr. Michael Zargham (@mzargham)",
            auto_attest=True,
            model_adequacy=_adequacy[_rid],
            evidence_sufficiency=_sufficiency[_rid],
        )

    mo.md("""
    ### Attestation outcomes

    | Requirement | Outcome | Reasoning |
    |---|---|---|
    | REQ-001 | `earl:failed`   | Settling time ~262s > 120s requirement; action item recorded |
    | REQ-002 | `earl:passed`   | Peak momentum well within 4.0 N.m.s |
    | REQ-003 | `earl:passed`   | Routh-Hurwitz parametric stability proof |
    | REQ-004 | `earl:passed`   | Gravity gradient torques 4 orders below actuator capacity |

    All four attestations are well-formed: each carries an adequacy
    `gsn:Assumption`, a sufficiency `gsn:Justification`, an EARL outcome,
    a qualified association naming the engineer's role and the procedure
    followed, and references to the evidence consulted. REQ-001 is
    *attested with `earl:failed`* — the audit graph records the gap
    explicitly rather than hiding it as "missing."
    """)
    return request_attestation


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 6: Closure-Rule Validation + Audit (initial)

    Before the program's chief systems engineer reviews the work, two
    automated checks ratify that the RTM graph itself is well-formed and
    internally consistent:

    1. **Closure-rule suite (SHACL).** Ten machine-checkable invariants —
       every attestation has both an adequacy Assumption and a sufficiency
       Justification, every evidence artifact has hashes and references a
       requirement, every analysis activity has an associated agent, etc.
       Plus a runtime re-verification check that re-hashes every proof.
    2. **Forward / Backward / Bidirectional audit.** Forward and backward
       run *independently* so the failure mode names which direction
       broke. Bidirectional is the derived conjunction.

    If any closure rule fails, the audit module's "fresh graph" claim
    can't be trusted. The two checks together are what an auditor would
    run before asking any substantive question.
    """)
    return


@app.cell(hide_code=True)
def __(rtm_graph, mo):
    from traceability.verification import verify as _verify

    _report = _verify(rtm_graph, skip_reverification=False)

    _summary = "\n".join("    " + l for l in _report.summary_lines())
    mo.md(
        "### Closure-rule suite (Stage 6.5)\n\n"
        "```\n"
        f"{_summary}\n"
        "```\n\n"
        "Ten invariants enforced — nine SHACL shapes + one runtime "
        "re-verification check. The shapes target distinct layers of the "
        "graph: attestation well-formedness, plan-instantiation correctness, "
        "evidence completeness, requirement structure, GSN argument well-"
        "formedness, PROV provenance shape, outcome semantics, "
        "forward/backward traceability, and named-graph integrity."
    )
    return


@app.cell(hide_code=True)
def __(rtm_graph, mo):
    from traceability.audit import audit as _audit_fn, render_report as _render

    audit_report = _audit_fn(rtm_graph)

    mo.md(
        "### Audit (Stage 7a)\n\n"
        "```\n"
        f"    {audit_report.forward.summary()}\n"
        f"    {audit_report.backward.summary()}\n"
        f"    Bidirectional: {'PASS' if audit_report.bidirectional().passed else 'FAIL'}\n"
        f"    Orphans: {'none' if not audit_report.orphans.any else 'see report'}\n"
        "```\n\n"
        "**Two orthogonal questions, not one.** The audit decouples\n\n"
        "- **traceability** — *is the structural chain intact?* "
        "Forward asks if every requirement is reached by evidence + an "
        "attestation; backward asks if every attestation's evidence "
        "actually addresses its claimed requirement. Bidirectional is "
        "the conjunction. These are structural facts about the graph.\n"
        "- **coverage status** — *is the requirement satisfied?* "
        "Outcome-derived per cell of the coverage matrix: "
        "`covered+passed` / `covered+failed` / `covered+cantTell` / "
        "`uncovered`. These are the engineering verdicts captured in "
        "each attestation's `earl:outcome`.\n\n"
        "**The v1 verdict (this stage).** Traceability passes — the "
        "chain is structurally sound, no orphans, every attestation "
        "lines up with its evidence. **REQ-001 is `covered+failed` and "
        "this is the *expected* output of v1**: the engineering analysis "
        "produced a real finding (settling time > spec) and the audit "
        "surfaces it as a coverage gap rather than hiding it behind "
        "'unattested'. Act 8 — Design Iteration — is the response, and "
        "Act 10's audit shows the same matrix flipping to all "
        "`covered+passed`."
    )
    return audit_report, _render


@app.cell(hide_code=True)
def __(audit_report, mo):
    _rows = "\n".join(
        f"| {c.requirement} | {c.evidence} | {c.status} |"
        for c in audit_report.coverage
    )
    mo.md(
        "**Coverage matrix**\n\n"
        "| Requirement | Evidence | Status |\n"
        "|---|---|---|\n"
        + _rows
    )
    return


@app.cell(hide_code=True)
def __(rtm_graph, mo):
    from interrogate.explain import explain_requirement

    explanations = {}
    for _rid in ["REQ-001", "REQ-002", "REQ-003", "REQ-004"]:
        explanations[_rid] = explain_requirement(rtm_graph, _rid)
    return explanations, explain_requirement


@app.cell(hide_code=True)
def __(explanations, mo):
    mo.md(f"""
    ### "How do you know REQ-003 is satisfied?"

    ```
    {explanations["REQ-003"]}
    ```

    The proof was **re-executed live** during this interrogation. The auditor
    doesn't need to trust the original analyst — they can see each lemma
    verified independently, right now.
    """)
    return


@app.cell(hide_code=True)
def __(explanations, mo):
    mo.md(f"""
    ### "What does the audit say about REQ-001?"

    ```
    {explanations["REQ-001"]}
    ```

    REQ-001 is **`covered+failed` at v1** — the structural trace from
    requirement → evidence → attestation is complete (which is why
    forward/backward both PASS), and the engineer recorded an
    `earl:failed` outcome with a precise reason (settling time > spec)
    and an action item (retune gains).

    This is the lifecycle state the demo is *designed* to produce at
    v1. The auditor sees three things separately:

    1. **The trace is intact.** No "missing attestation", no orphan
       evidence — the graph is internally consistent.
    2. **The engineering verdict is recorded.** Outcome = `earl:failed`,
       with adequacy and sufficiency text on linked GSN nodes.
    3. **The next action is in the record.** The sufficiency
       `gsn:Justification` carries the corrective recommendation, so
       the design iteration in Act 8 has a documented starting point.
    """)
    return


@app.cell(hide_code=True)
def __(rtm_graph, mo):
    from interrogate.reproduce import reproduce_all_evidence

    _repro = reproduce_all_evidence(rtm_graph)

    _proof_rows = []
    for _p in _repro["proofs"]:
        _match = "MATCH" if _p["hash_match"] else "MISMATCH"
        _proof_rows.append(f"| {_p['requirement']} | {_p['status'].value} | {_match} |")

    _repro_table = "\n".join(_proof_rows)
    _n_sims = len(_repro['simulations'])
    mo.md(
        "### Reproducibility Audit\n\n"
        "The auditor re-executes ALL computational evidence:\n\n"
        "**Proof Re-verification:**\n\n"
        "| Requirement | Status | Hash Match |\n"
        "|-------------|--------|-----------|\n"
        f"{_repro_table}\n\n"
        f"**Simulation Reproduction:** {_n_sims} simulations re-run successfully.\n\n"
        "Every proof re-verifies. Every hash matches. The evidence is "
        "reproducible — not because we say so, but because the auditor "
        "just confirmed it."
    )
    return reproduce_all_evidence


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 7: The Traceability Graph

    The complete v1 RTM as a directed graph. Requirements (blue) flow
    through design elements (green) to evidence (orange/yellow) to
    attestations (red). Every edge is a queryable RDF triple in git.

    What you should see at v1:

    - **All four requirements have full chains** to evidence and an
      attestation node — that's why Act 6's audit reported the trace
      as PASS. The chain is structurally complete for every
      requirement.
    - **REQ-001's attestation carries `earl:failed`** in the outcome —
      the engineering verdict — even though its trace looks just like
      the other three at the graph level. Outcome lives on the
      attestation node as a property, not in the link shape.

    The `print_rtm_summary` block below makes this distinction
    explicit: requirements are categorized as ATTESTED with their
    EARL outcome surfaced, separating "trace intact + verdict pending
    work" from "trace incomplete."
    """)
    return


@app.cell(hide_code=True)
def __(rtm_graph):
    from interrogate.visualize import build_rtm_figure

    rtm_fig = build_rtm_figure(rtm_graph, figsize=(18, 10))
    rtm_fig
    return rtm_fig, build_rtm_figure


@app.cell(hide_code=True)
def __(rtm_graph, mo):
    from traceability.rtm import print_rtm_summary

    _summary = print_rtm_summary(rtm_graph)
    mo.md(f"""
    ### Final Status

    ```
    {_summary}
    ```
    """)
    return print_rtm_summary


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 8: Design Iteration

    The open finding on REQ-001 drives action. We need to increase the
    derivative gain Kd to reduce settling time. Rather than editing a file
    by hand, we apply a **SPARQL UPDATE** to the structural model — the
    same way any RDF-native toolchain would propagate a design change.

    This demonstrates **reproducibility as regression testing**: the model
    hash changes, all previous proofs are invalidated, and we must re-run
    the entire analysis chain from scratch.
    """)
    return


@app.cell(hide_code=True)
def __(struct_graph, mo):
    from rdflib import Literal, XSD

    # Clone the graph for the design iteration
    from rdflib import Graph as _G
    from ontology.prefixes import bind_prefixes as _bp, SYSML as _SYSML

    v2_graph = _G()
    _bp(v2_graph)
    for triple in struct_graph:
        v2_graph.add(triple)

    # Apply SPARQL UPDATE: retune both controller gains
    # Increasing Kd alone would overdamp the system — a controls engineer
    # retunes both Kp (stiffness) and Kd (damping) together.
    _sparql_update_kd = """
    DELETE { ?attr sysml:value ?oldVal }
    INSERT { ?attr sysml:value "30.0"^^xsd:double }
    WHERE {
        ?attr sysml:declaredName "Kd" ;
              sysml:value ?oldVal .
    }
    """
    _sparql_update_kp = """
    DELETE { ?attr sysml:value ?oldVal }
    INSERT { ?attr sysml:value "4.0"^^xsd:double }
    WHERE {
        ?attr sysml:declaredName "Kp" ;
              sysml:value ?oldVal .
    }
    """
    v2_graph.update(_sparql_update_kd, initNs={"sysml": _SYSML})
    v2_graph.update(_sparql_update_kp, initNs={"sysml": _SYSML})

    # Verify
    _check = """
    SELECT ?name ?val WHERE {
        ?attr sysml:declaredName ?name ;
              sysml:value ?val .
        FILTER(?name IN ("Kp", "Kd"))
    }
    ORDER BY ?name
    """
    _gains = {str(r[0]): float(r[1]) for r in v2_graph.query(_check, initNs={"sysml": _SYSML})}

    mo.md(
        "### SPARQL UPDATE: Controller Retune\n\n"
        "A controls engineer retunes both gains together — increasing Kd alone\n"
        "would overdamp the system, making settling *slower* despite more damping.\n\n"
        "```sparql\n"
        "# Increase proportional gain (stiffness)\n"
        'DELETE { ?attr sysml:value ?oldVal }\n'
        'INSERT { ?attr sysml:value "4.0"^^xsd:double }\n'
        'WHERE  { ?attr sysml:declaredName "Kp" ; sysml:value ?oldVal . }\n\n'
        "# Increase derivative gain (damping)\n"
        'DELETE { ?attr sysml:value ?oldVal }\n'
        'INSERT { ?attr sysml:value "30.0"^^xsd:double }\n'
        'WHERE  { ?attr sysml:declaredName "Kd" ; sysml:value ?oldVal . }\n'
        "```\n\n"
        f"| Gain | Before | After |\n"
        f"|------|--------|-------|\n"
        f"| Kp | 1.0 N.m/rad | **{_gains['Kp']:.1f}** N.m/rad |\n"
        f"| Kd | 10.0 N.m.s/rad | **{_gains['Kd']:.1f}** N.m.s/rad |"
    )
    return v2_graph


@app.cell(hide_code=True)
def __(v2_graph, model_hash, mo):
    from evidence.hashing import hash_structural_model as _hsm

    v2_model_hash = _hsm(v2_graph)

    mo.md(
        "### Model Hash Invalidation\n\n"
        f"| | Hash |\n"
        f"|--|------|\n"
        f"| Original model | `{model_hash[:24]}...` |\n"
        f"| Updated model  | `{v2_model_hash[:24]}...` |\n\n"
        "The hashes differ — **all previous proofs and evidence are now invalid**.\n"
        "Any attempt to verify an old proof against the new model hash will fail.\n"
        "We must re-derive everything from scratch."
    )
    return v2_model_hash


@app.cell(hide_code=True)
def __(v2_graph, v2_model_hash, mo):
    from analysis.load_params import load_params as _lp
    from analysis.symbolic import run_symbolic_analysis as _rsa
    from analysis.build_proofs import build_all_proofs as _bap
    from analysis.proof_scripts import verify_proof as _vp, ProofStatus as _PS

    v2_params = _lp(v2_graph)
    v2_sym = _rsa(v2_params)
    v2_margins = v2_sym.stability_margins

    v2_proofs = _bap(v2_model_hash)
    v2_proof_results = {}
    for _rid, _script in v2_proofs.items():
        v2_proof_results[_rid] = _vp(_script, v2_model_hash)

    _margin_rows = "\n".join(
        f"| {axis} | {val:.4f} | {'PASS' if val <= -0.010 else 'FAIL'} |"
        for axis, val in v2_margins.items()
    )
    _proof_rows = "\n".join(
        f"| {rid} | {'VERIFIED' if r.status == _PS.VERIFIED else 'FAILED'} |"
        for rid, r in sorted(v2_proof_results.items())
    )

    mo.md(
        "### Re-run Symbolic Analysis (Kp=4, Kd=30)\n\n"
        f"Inertia unchanged: Jxx={v2_sym.inertia[0]:.1f}, "
        f"Jyy={v2_sym.inertia[1]:.1f}, Jzz={v2_sym.inertia[2]:.1f} kg.m^2\n\n"
        "**Stability margins (improved):**\n\n"
        "| Axis | Re(lambda) | Status |\n"
        "|------|-----------|--------|\n"
        f"{_margin_rows}\n\n"
        f"Settling time estimate: **{v2_sym.pointing_budget['settling_time_s']:.1f}s** "
        f"(was 262s, requirement is 120s)\n\n"
        "**Proofs (re-verified against new model hash):**\n\n"
        "| Requirement | Status |\n"
        "|-------------|--------|\n"
        f"{_proof_rows}"
    )
    return v2_params, v2_sym, v2_proofs, v2_proof_results


@app.cell(hide_code=True)
def __(v2_params, mo):
    from analysis.numerical import run_step_response as _rsr

    v2_step = _rsr(v2_params)
    v2_step_summary = v2_step.summary()

    import matplotlib.pyplot as _plt2
    import numpy as _np2

    _fig2, _ax2 = _plt2.subplots(1, 2, figsize=(14, 5))
    _axis_colors = {"X": "#1f77b4", "Y": "#2ca02c", "Z": "#9467bd"}
    _limit_color = "#d62728"

    # Pointing convergence comparison
    _q_vec2 = _np2.linalg.norm(v2_step.q[:, :3], axis=1)
    _theta2 = _np2.degrees(2 * _q_vec2)
    _ax2[0].semilogy(v2_step.t, _theta2, color=_axis_colors["X"], linewidth=1.5, label="Kp=4, Kd=30 (updated)")
    _ax2[0].axhline(0.1, color=_limit_color, linestyle="--", linewidth=1, label="REQ-001 limit")
    _ax2[0].axvline(120, color="#888", linestyle=":", linewidth=1, alpha=0.7, label="120s target")
    _ax2[0].set_xlabel("Time (s)")
    _ax2[0].set_ylabel("Attitude Error (deg)")
    _ax2[0].set_title("Pointing Convergence (Kp=4, Kd=30)")
    _ax2[0].set_ylim(bottom=1e-4)
    _ax2[0].legend(fontsize=8)
    _ax2[0].grid(True, alpha=0.3, which="both")

    # Wheel momentum
    _h2 = _np2.linalg.norm(v2_step.h_wheel, axis=1)
    _ax2[1].plot(v2_step.t, _h2, color=_axis_colors["X"], linewidth=1.5, label="|h| (total)")
    _ax2[1].axhline(v2_step.config.max_momentum, color=_limit_color, linestyle="--", linewidth=1, label="REQ-002 limit")
    _ax2[1].set_xlabel("Time (s)")
    _ax2[1].set_ylabel("Momentum (N.m.s)")
    _ax2[1].set_title("Wheel Momentum (Kp=4, Kd=30)")
    _ax2[1].legend(fontsize=8)
    _ax2[1].grid(True, alpha=0.3)

    _fig2.suptitle("Design Iteration: Step Response with Kp=4, Kd=30", fontsize=13, fontweight="bold")
    _plt2.tight_layout()

    mo.md(
        "### Re-run Numerical Simulation (Kp=4, Kd=30)\n\n"
        f"| Metric | Kd=10 (original) | Kd=30 (updated) | Requirement |\n"
        f"|--------|-----------------|-----------------|-------------|\n"
        f"| Settling time | 262s | **{v2_step_summary['settling_time_s']:.1f}s** | < 120s |\n"
        f"| Final error | 0.1223 deg | **{v2_step_summary['final_error_deg']:.4f} deg** | < 0.1 deg |\n"
        f"| Peak momentum | 0.810 N.m.s | **{v2_step_summary['peak_wheel_momentum']:.3f} N.m.s** | < 4.0 N.m.s |\n"
        f"| Peak torque | 0.044 N.m | **{v2_step_summary['peak_control_torque']:.4f} N.m** | < 0.1 N.m |\n"
    )

    _fig2
    return v2_step, v2_step_summary


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ### Regression Check

    With Kp=4, Kd=30, the settling time is now well under 120s. But we must
    verify that the gain increase didn't break anything else:

    - **REQ-002 (momentum):** Peak momentum may increase (higher gains → more
      aggressive control), but must remain within the 4.0 N.m.s limit.
    - **REQ-003 (stability):** Higher gains improve both bandwidth and damping.
    - **REQ-004 (disturbance):** Gravity gradient rejection improves — higher Kp
      reduces steady-state error from disturbances.

    This is **reproducibility as regression testing**. The same pipeline that found
    the deficiency now confirms the fix doesn't introduce new problems.
    """)
    return


@app.cell(hide_code=True)
def __(v2_graph, v2_model_hash, v2_params, v2_proofs, v2_step_summary, v2_step, mo):
    from evidence.binding import (
        bind_proof_evidence as _bpe, bind_simulation_evidence as _bse,
        bind_computation_engines as _bce,
    )
    from evidence.hashing import (
        hash_proof as _hp2, hash_evidence as _he2, hash_simulation as _hs2,
    )
    from pipeline.dataset import graph_for as _gf
    from traceability.rtm import load_base_dataset as _lbd
    from traceability.attestation import request_attestation as _ra

    # Build v2 RTM as a Dataset so the new audit module (Act 10) can
    # query it with named-graph awareness. Replace the structural layer
    # with the v2 model (Kp=4, Kd=30).
    v2_rtm = _lbd()
    _struct_v2 = _gf(v2_rtm, "structural")
    # Wipe the original structural triples and replace with v2_graph
    for _t in list(_struct_v2):
        _struct_v2.remove(_t)
    for _t in v2_graph:
        _struct_v2.add(_t)

    _ev2 = _gf(v2_rtm, "evidence")
    _bce(_ev2)

    for _rid, _script in v2_proofs.items():
        _ph = _hp2(_script, v2_model_hash)
        _ch = _he2(v2_model_hash, proof_hash=_ph)
        _bpe(_ev2, f"EV-PROOF-{_rid}-v2", f"SA-{_rid}-v2", _rid,
             v2_model_hash, _ph, _ch,
             f"Symbolic proof (v2): {_script.claim}",
             source_file="analysis/build_proofs.py")

    _sh2 = _hs2(v2_step.config.to_dict(), v2_step_summary)
    for _rid, _desc in [
        ("REQ-001", f"Step response (v2): settling={v2_step_summary['settling_time_s']:.1f}s, "
                    f"final_error={v2_step_summary['final_error_deg']:.4f} deg"),
        ("REQ-002", f"Peak wheel momentum (v2): {v2_step_summary['peak_wheel_momentum']:.3f} N.m.s"),
    ]:
        _bse(_ev2, f"EV-SIM-{_rid}-v2", f"NS-{_rid}-v2", _rid,
             v2_model_hash, _sh2, _desc, source_file="analysis/numerical.py")

    # Attest ALL 4 requirements with earl:passed — the v2 evidence is
    # now sufficient (settling time met).
    _v2_adequacy = {
        "REQ-001": ("Linearized PD model adequate. Kp increased to 4, Kd to 30 per design review finding. "
                    "Interface parameters unchanged from systems engineering."),
        "REQ-002": "Energy-based momentum bound conservative. Peak may increase with higher gains but within limits.",
        "REQ-003": "Routh-Hurwitz stability confirmed. Higher gains improve damping and bandwidth.",
        "REQ-004": "GG model adequate. Controller gain change does not affect disturbance torque magnitude.",
    }
    _v2_sufficiency = {
        "REQ-001": (f"Settling time now {v2_step_summary['settling_time_s']:.1f}s (< 120s requirement). "
                    f"Pointing accuracy {v2_step_summary['final_error_deg']:.4f} deg (< 0.1 deg). Both met."),
        "REQ-002": f"Peak momentum {v2_step_summary['peak_wheel_momentum']:.3f} N.m.s, well within 4.0 limit.",
        "REQ-003": "Routh-Hurwitz proof valid for all positive J, Kp, Kd. Margins improved with higher gains.",
        "REQ-004": "GG torques unchanged at ~1e-6 N.m. Overwhelming margin vs 0.1 N.m actuator capacity.",
    }

    for _rid in ["REQ-001", "REQ-002", "REQ-003", "REQ-004"]:
        _ra(v2_rtm, _rid, "Dr. Michael Zargham (@mzargham)",
            auto_attest=True,
            model_adequacy=_v2_adequacy[_rid],
            evidence_sufficiency=_v2_sufficiency[_rid])

    mo.md("""
    ### Full Re-attestation (v2 model)

    With the updated model, all four requirements now have sufficient evidence:

    - **REQ-001: ATTESTED** — settling time and pointing accuracy both met
    - **REQ-002: ATTESTED** — momentum within limits (slight increase noted)
    - **REQ-003: ATTESTED** — stability margins improved
    - **REQ-004: ATTESTED** — disturbance rejection unchanged

    The entire evidence chain is fresh — new model hash, new proof hashes,
    new simulation hashes. Nothing carries over from the v1 analysis.
    """)
    return v2_rtm


@app.cell(hide_code=True)
def __(v2_rtm):
    from interrogate.visualize import build_rtm_figure as _brf

    v2_fig = _brf(v2_rtm, figsize=(18, 10), title="Requirements Traceability Matrix (v2: Kp=4, Kd=30)")
    v2_fig
    return v2_fig


@app.cell(hide_code=True)
def __(v2_rtm, mo):
    from interrogate.explain import explain_requirement as _er

    _v2_expl = _er(v2_rtm, "REQ-001")
    mo.md(
        '### "How do you know REQ-001 is satisfied now?"\n\n'
        f"```\n{_v2_expl}\n```\n\n"
        "The traceability chain is now **complete** for REQ-001. The auditor can see:\n\n"
        "- Fresh proof artifact bound to the v2 model hash\n"
        "- Fresh simulation showing settling time under 120s\n"
        "- Attestation by Dr. Zargham with explicit reference to the design change\n"
        "- The v2 model hash differs from v1 — confirming this is new evidence, not recycled"
    )
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 9: Remote Compute & Distribution

    Up to now the analysis ran on the engineer's local machine. In
    production aerospace deployments, the **compute server** is usually a
    different physical host than where the engineer reviews results — a
    remote analysis cluster, a CI runner, or a containerized environment
    pinned to a known toolchain. The RTM must record *where* and *how*
    each piece of evidence was produced, otherwise the audit chain has
    a gap.

    Two pluggable extensions:

    1. **Compute backends** — `LocalCompute` (default, in-process) or
       `DockerCompute` (ephemeral container per analysis stage). The
       container's identity (image digest, container ID, hostname) is
       captured as PROV-O triples on the analysis activity.
    2. **Persistence backends** — `LocalBackend` (filesystem) or
       `FlexoBackend` / `FuskeiBackend` (a real quadstore). Each named
       graph in our `rdflib.Dataset` maps to a Flexo branch.

    For this notebook we annotate v2's analysis activities *as if* they
    had run inside an `adcs-compute:latest` container — so the auditor
    can see the captured execution context in Act 10. In a live run
    `--compute=docker` does the actual container spawn and captures
    real digests.
    """)
    return


@app.cell(hide_code=True)
def __(v2_rtm, mo):
    from compute.base import ExecutionMetadata as _ExecMeta
    from evidence.binding import _bind_execution_metadata as _bind_meta
    from ontology.prefixes import ADCS as _ADCS, G_EVIDENCE as _G_EV
    from rdflib import URIRef as _URI

    # Simulate captured metadata from a Docker compute run. In a live
    # pipeline (`--compute=docker`) these values come from the real
    # `docker image inspect` + container hostname + cidfile.
    _exec_meta = _ExecMeta(
        location_kind="docker",
        hostname="container-71a59f23f3e9",
        image_digest="sha256:92bb8bf18f5f2ba7a6e332e4fe1fa1b12911e9b6c4cddb4b35e1659b01b21d30",
        image_label="adcs-compute:latest",
        container_id="71a59f23f3e9",
        python_version="3.12.13",
        started_at="2026-05-14T02:27:51+00:00",
        ended_at="2026-05-14T02:27:56+00:00",
    )

    _ev_g = v2_rtm.graph(_URI(_G_EV))
    for _rid in ["REQ-001", "REQ-002", "REQ-003", "REQ-004"]:
        _bind_meta(_ev_g, _ADCS[f"SA-{_rid}-v2"], _exec_meta)
        _bind_meta(_ev_g, _ADCS[f"NS-{_rid}-v2"], _exec_meta)

    mo.md(
        "### Remote-compute provenance attached\n\n"
        "Each v2 analysis activity (`SA-*-v2`, `NS-*-v2`) now carries:\n\n"
        "```turtle\n"
        "adcs:SA-REQ-003-v2\n"
        "    prov:atLocation        <urn:adcs:location:docker:container-71a59f23f3e9> ;\n"
        "    prov:wasAssociatedWith <urn:adcs:executor:71a59f23f3e9> ;\n"
        "    prov:startedAtTime     \"2026-05-14T02:27:51+00:00\"^^xsd:dateTime ;\n"
        "    prov:endedAtTime       \"2026-05-14T02:27:56+00:00\"^^xsd:dateTime .\n\n"
        "<urn:adcs:executor:71a59f23f3e9> a prov:SoftwareAgent ;\n"
        "    rtm:hostname      \"container-71a59f23f3e9\" ;\n"
        "    rtm:imageDigest   \"sha256:92bb8bf18f5f...\" ;\n"
        "    rtm:imageLabel    \"adcs-compute:latest\" ;\n"
        "    rtm:containerId   \"71a59f23f3e9\" ;\n"
        "    rtm:pythonVersion \"3.12.13\" .\n"
        "```\n\n"
        "The image digest pins the toolchain version cryptographically. "
        "If someone replays the analysis, they pull the same image by "
        "digest and get an identical environment — not 'a Python with "
        "scipy somewhere' but *that* Python with *that* scipy.\n\n"
        "**Distribution** is the parallel story for persistence. The same "
        "v2_rtm Dataset can be pushed to a real quadstore:\n\n"
        "```bash\n"
        "export FLEXO_TOKEN=...     # from a collaborator on the sandbox\n"
        "make flexo-init             # one-time: provision org / repo / master\n"
        "make flexo-run              # = pipeline.runner --backend=flexo\n"
        "```\n\n"
        "Live result against `try-layer1.starforge.app`:\n\n"
        "| Branch | Triples |\n"
        "|---|---:|\n"
        "| ontology | 317 |\n"
        "| structural | 253 |\n"
        "| evidence | 126 |\n"
        "| attestations | 89 |\n"
        "| plan-execution | 52 |\n"
        "| audit | 8 |\n\n"
        "`SPARQL ASK { adcs:ATT-REQ-003 a rtm:Attestation }` returns "
        "`true` on the attestations branch — the data is queryable end-"
        "to-end from a host the engineer never logged into."
    )
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ### Many Authoritative Sources of Truth, one stitched provenance graph

    The RTM you've been browsing is **not** a single authoritative store
    pretending its claims are first-hand. It's a graph stitched out of
    many **Authoritative Sources of Truth (ASoTs)**, each holding part
    of the picture under its own jurisdiction. The demo's contribution
    is the integration ontology that lets them link without overloading
    anyone's vocabulary.

    | ASoT | Holds | Trust grounded by |
    |---|---|---|
    | **SysMLv2 structural model** (git) | requirements + design elements + satisfy links | `sysml:declaredName`, vendored OMG SysMLv2 OWL equivalences |
    | **Symbolic oracle** (SymPy in container) | analytical proofs (stability, momentum bounds) | content-addressed `rtm:proofHash` over the proof script |
    | **Numerical oracle** (scipy in container) | step-response + disturbance-rejection simulations | content-addressed `rtm:contentHash` over (config, summary) |
    | **Docker image** (built from git Dockerfile) | runtime substrate of the oracles | `rtm:contentHash` (runtime digest) + `rtm:dockerfileHash` + `rtm:buildContextHash` + `rtm:gitRef` |
    | **Engineer attestation** (human judgement) | adequacy + sufficiency for each requirement | `gsn:Assumption` + `gsn:Justification`, signed via `rtm:attests` |
    | **Closure-rule check** (SHACL) | every-run conformance to the integration shapes | `rtm:ClosureRuleAssertion` (`earl:Assertion`, `earl:mode earl:automatic`) |
    | **Audit module** (forward + backward graph traversal) | trace-matrix closure verdicts | computed deterministically; persisted to `<adcs:audit>` |

    Each oracle is **content-addressed at its output** and
    **provenance-linked at its execution context** — so when the audit
    asks "how can I trust this evidence?" the answer walks
    `evidence → wasGeneratedBy → activity → used → container → wasDerivedFrom → image → gitRef`
    without leaving the graph. The shape catalog (`rtm:ProofArtifact`,
    `rtm:SimulationResult`, `rtm:DockerImage`, `rtm:DockerContainer`)
    is the integration ontology's whole inventory — every term is a
    subclass or property of something upstream (PROV-O, EARL, OntoGSN,
    SHACL); nothing new on the epistemic axis.
    """)
    return


@app.cell(hide_code=True)
def __(v2_rtm, mo):
    # Synthesize the Docker image record + cross-links a live
    # --compute=docker --backend=flexo run would emit, so the
    # downstream evidence_by_image query and the numerical-evidence
    # provenance render have data to walk.
    from rdflib import Literal as _Lit, URIRef as _URI
    from rdflib.namespace import RDF as _RDF, XSD as _XSD
    from ontology.prefixes import (
        ADCS as _ADCS, G_EVIDENCE as _G_EV,
        PROV as _PROV, RTM as _RTM,
    )

    _ev_g = v2_rtm.graph(_URI(_G_EV))
    _image = _URI("urn:adcs:docker-image:sha256-92bb8bf18f5f2ba7a6e332e4fe1fa1b12911e9b6c4cddb4b35e1659b01b21d30")
    _ev_g.add((_image, _RDF.type, _RTM.DockerImage))
    _ev_g.add((_image, _RDF.type, _PROV.Entity))
    _ev_g.add((_image, _RTM.contentHash, _Lit("sha256:92bb8bf18f5f2ba7a6e332e4fe1fa1b12911e9b6c4cddb4b35e1659b01b21d30")))
    _ev_g.add((_image, _RTM.imageLabel, _Lit("adcs-compute:latest")))
    _ev_g.add((_image, _RTM.dockerfileHash, _Lit("sha256:dockerfile-bytes-hash-here")))
    _ev_g.add((_image, _RTM.buildContextHash, _Lit("sha256:build-context-manifest-hash")))
    _ev_g.add((_image, _RTM.gitRef,
               _Lit("git+https://github.com/DynamicalSystemsGroup/ADCS-lifecycle-demo@HEAD#compute/Dockerfile",
                    datatype=_XSD.anyURI)))
    _ev_g.add((_image, _RTM.flexoRecord,
               _URI("urn:adcs:flexo:adcs-demo/lifecycle/evidence")))

    # Container materialization (per-run, content-addressed by container id)
    _container = _URI("urn:adcs:docker-container:71a59f23f3e9")
    _ev_g.add((_container, _RDF.type, _RTM.DockerContainer))
    _ev_g.add((_container, _RDF.type, _PROV.Entity))
    _ev_g.add((_container, _RTM.containerId, _Lit("71a59f23f3e9")))
    _ev_g.add((_container, _PROV.wasDerivedFrom, _image))

    # Link every v2 analysis activity to the container (the runtime
    # it actually executed in) and tag each piece of v2 evidence as
    # derived from the image (the recipe).
    for _rid in ["REQ-001", "REQ-002", "REQ-003", "REQ-004"]:
        _ev_g.add((_ADCS[f"SA-{_rid}-v2"], _PROV.used, _container))
        _ev_g.add((_ADCS[f"NS-{_rid}-v2"], _PROV.used, _container))
        _ev_g.add((_ADCS[f"EV-PROOF-{_rid}-v2"], _PROV.wasDerivedFrom, _image))
        _ev_g.add((_ADCS[f"EV-SIM-{_rid}-v2"], _PROV.wasDerivedFrom, _image))

    # Inverse query: from the image, find every evidence node derived from it.
    from traceability.queries import evidence_by_image as _evbyimg
    _rows = _evbyimg(v2_rtm, "sha256:92bb8bf18f5f2ba7a6e332e4fe1fa1b12911e9b6c4cddb4b35e1659b01b21d30")
    _proof_count = sum(1 for r in _rows if "ProofArtifact" in (r.get("type") or ""))
    _sim_count = sum(1 for r in _rows if "SimulationResult" in (r.get("type") or ""))

    mo.md(
        "### The runtime ASoTs as first-class nodes\n\n"
        "The symbolic oracle and the numerical oracle both run inside an "
        "`adcs-compute:latest` container materialized from a Dockerfile "
        "tracked in git. The container's identity (recipe + runtime + git ref) "
        "is itself a recorded fact:\n\n"
        "```turtle\n"
        "<urn:adcs:docker-image:sha256-92bb8bf18f5f...>\n"
        "    a rtm:DockerImage, prov:Entity ;\n"
        "    rtm:contentHash       \"sha256:92bb8bf18f5f...\" ;     # runtime digest\n"
        "    rtm:imageLabel        \"adcs-compute:latest\" ;\n"
        "    rtm:dockerfileHash    \"sha256:dockerfile-bytes-...\" ; # the recipe\n"
        "    rtm:buildContextHash  \"sha256:build-context-...\" ;\n"
        "    rtm:gitRef            \"git+https://github.com/.../@HEAD#compute/Dockerfile\"^^xsd:anyURI ;\n"
        "    rtm:flexoRecord       <urn:adcs:flexo:adcs-demo/lifecycle/evidence> .\n\n"
        "<urn:adcs:docker-container:71a59f23f3e9>\n"
        "    a rtm:DockerContainer, prov:Entity ;\n"
        "    rtm:containerId        \"71a59f23f3e9\" ;\n"
        "    prov:wasDerivedFrom    <urn:adcs:docker-image:sha256-92bb...> .  # materialization\n\n"
        "adcs:NS-REQ-001-v2 prov:used <urn:adcs:docker-container:71a59f23f3e9> .\n"
        "adcs:EV-SIM-REQ-001-v2 prov:wasDerivedFrom <urn:adcs:docker-image:sha256-92bb...> .\n"
        "```\n\n"
        f"Image → evidence (the inverse query): **{_proof_count} proof artifact(s) + {_sim_count} simulation result(s)** "
        "derived from this exact image digest.\n\n"
        "A reviewer can ask:\n\n"
        "- *what code produced this evidence?* — walk `evidence → wasDerivedFrom → image → gitRef`\n"
        "- *which container ran it?* — walk `evidence → wasGeneratedBy → activity → used → container`\n"
        "- *can I reproduce it?* — `compute.reproduce --image-digest sha256:92bb...` clones the recorded git ref, rebuilds the image, and emits an `rtm:DigestMatchAssertion` (`earl:passed` / `earl:failed`) into the audit graph\n\n"
        "The recipe lives in git; the witness lives in Flexo; the runtime "
        "materializes locally; the wire-level invocation log can live in a "
        "fourth-service document store. Each piece is under its own auspices "
        "(`prov:wasAttributedTo` for operators, `rtm:operatedBy` for hosts) and "
        "the integration ontology stitches them via standard PROV edges."
    )
    return


@app.cell(hide_code=True)
def __(v2_rtm, mo):
    # Numerical-evidence provenance rendered as a concrete trace,
    # not abstract narration. Pick a single EV-SIM-* node and walk
    # every linked fact — readers see the full chain in one block
    # rather than having to query for it.
    from traceability.queries import (
        technical_provenance as _techprov,
        render_trust_summary as _render,
        trust_summary as _trust,
    )

    _focus = "http://example.org/adcs-demo/EV-SIM-REQ-001-v2"
    _summary = _trust(v2_rtm, _focus)
    _txt = _render(_summary)

    mo.md(
        "### Numerical evidence — the full provenance, end-to-end\n\n"
        "The simulation evidence node for REQ-001's step-response check "
        "(`EV-SIM-REQ-001-v2`) carries the chain a reviewer needs to ask "
        "every \"how can I trust this?\" question.\n\n"
        f"```text\n{_txt}\n```\n\n"
        "Read this top-to-bottom and you get: which oracle ran (Scipy "
        "numerical-simulation engine, executed as the analysis activity); "
        "which container materialized the oracle (`71a59f23f3e9`); which "
        "image's recipe produced that container (`sha256:92bb8bf18f5f...`); "
        "which git commit was the source-of-truth for the recipe; which "
        "host the container ran on; which organization operated the host; "
        "and the closure-rule outcome attesting the dataset still conforms "
        "to its SHACL invariants. **No fact requires trusting a single party** "
        "— each link is independently auditable against its own ASoT, and a "
        "reproducer can rebuild the image at the recorded git ref, re-run "
        "the simulation, and check the numerical output matches "
        "`rtm:contentHash`."
    )
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Act 10: Fresh Audit (after remote compute)

    The audit module is **backend-agnostic** and **compute-agnostic** —
    it queries the local Dataset whether the analysis ran in-process or
    in a container, whether the persistence lands on disk or in Flexo.
    What the auditor gets, after Act 9's annotations, is a richer trace
    that includes the execution context per activity.

    We re-run the audit against the v2 Dataset and surface two things
    the original audit (Act 6) couldn't show:

    1. The complete forward / backward / bidirectional verdict on v2
       (now that REQ-001 passes too).
    2. The execution-context triples — a SPARQL query that answers
       *"Where and on what image was each piece of v2 evidence
       produced?"*
    """)
    return


@app.cell(hide_code=True)
def __(v2_rtm, mo):
    from traceability.audit import audit as _audit_fn_v2

    _v2_audit = _audit_fn_v2(v2_rtm)
    _rows = "\n".join(
        f"| {c.requirement} | {c.evidence} | {c.status} |"
        for c in _v2_audit.coverage
    )
    mo.md(
        "### v2 audit\n\n"
        "```\n"
        f"    {_v2_audit.forward.summary()}\n"
        f"    {_v2_audit.backward.summary()}\n"
        f"    Bidirectional: {'PASS' if _v2_audit.bidirectional().passed else 'FAIL'}\n"
        f"    Orphans: {'none' if not _v2_audit.orphans.any else 'see report'}\n"
        "```\n\n"
        "**Coverage matrix (v2):**\n\n"
        "| Requirement | Evidence | Status |\n"
        "|---|---|---|\n"
        + _rows + "\n\n"
        "Every requirement now shows `covered+passed`. The same audit "
        "code that flagged REQ-001 as `covered+failed` in Act 6 here "
        "reports it as resolved — and the closure-rule suite, "
        "explanation chain, and re-verification all continue to pass "
        "(the design iteration didn't break anything else)."
    )
    return


@app.cell(hide_code=True)
def __(v2_rtm, mo):
    # SPARQL across the union to pull each analysis activity's execution
    # context. default_union makes this work without explicit GRAPH
    # clauses — see Prologue.
    _q = """
    PREFIX prov: <http://www.w3.org/ns/prov#>
    PREFIX rtm:  <http://example.org/ontology/rtm#>
    SELECT ?activity ?location ?image ?host ?started WHERE {
        ?activity prov:atLocation ?location ;
                  prov:wasAssociatedWith ?executor ;
                  prov:startedAtTime ?started .
        ?executor a prov:SoftwareAgent ;
                  rtm:imageLabel ?image ;
                  rtm:hostname ?host .
    }
    ORDER BY ?activity
    """
    _rows = []
    for r in v2_rtm.query(_q):
        _act = str(r.activity).rsplit("/", 1)[-1]
        _img = str(r.image)
        _host = str(r.host)
        _rows.append(f"| {_act} | {_img} | {_host} |")

    if not _rows:
        _body = "_no execution-context triples found_"
    else:
        _body = (
            "| Activity | Image | Host |\n"
            "|---|---|---|\n"
            + "\n".join(_rows)
        )

    mo.md(
        '### "Where and on what image was each v2 piece of evidence produced?"\n\n'
        f"{_body}\n\n"
        "The audit chain now goes all the way to the executor. An auditor "
        "reviewing this RTM can pull the exact image by digest, replay "
        "the analysis, and verify the proofs re-derive bit-for-bit — "
        "without trusting the engineer's local environment."
    )
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Future Work

    The demo deliberately stops short of several production-grade
    extensions, each documented in
    [`/Users/z/.claude/plans/i-want-to-look-hidden-balloon.md`](#) and
    summarized below:

    - **Cryptographic envelopes & signatures.** Today hashes are bare
      SHA-256: content identity, but not authenticity. Production
      should layer W3C VC Data Integrity (Ed25519 over RDF
      canonicalization), in-toto/SLSA build attestations, and
      sigstore/Rekor transparency logs.
    - **Formal authority & credential model.** FOAF + W3C Org Ontology +
      `schema:hasCredential` + W3C Verifiable Credentials on top of
      `prov:Agent` — so an attestation can carry not just *who*
      attested but *what role they were authorized in* and *what
      credentials backed that authorization*.
    - **OntoGSN confidence arguments.** Reify confidence in each
      Assumption / Justification node so stakeholders can ask "how
      confident are you in your adequacy claim?" as a queryable graph
      instead of prose.
    - **Defeaters & revocation.** SACM/OntoGSN-style invalidation of
      attestations when later evidence contradicts an earlier
      assumption (e.g., test-flight data invalidates a linearization
      regime assumption).
    - **Multi-attestation aggregation.** Sign-off policies (Engineering
      + QA + Certifier must all attest with `earl:passed`) expressed
      as SHACL gates on requirement transitions.
    - **Production Flexo deployment.** Multi-user auth (SSO),
      PR-style branches for RTM evolution, CI hooks, federation across
      program-level Flexo instances.
    - **OSLC connector** for DOORS Next / Jama / RQM.
    - **Federated SPARQL** for cross-program traceability.
    - **Continuous re-verification in CI** and a **live traceability
      dashboard** — both compose with the cryptographic-envelope work.
    """)
    return


@app.cell(hide_code=True)
def __(mo):
    mo.md("""
    ---

    ## Summary

    The complete ADCS verification lifecycle, numbered to the Acts so it
    doubles as a table of contents. Sub-indexes (a/b/c) call out
    multiple distinct actions inside one Act.

    **Prologue.** Assembled the integration ontology from PROV-O + EARL
    + OntoGSN + P-PLAN + OSLC RM/QM + the OMG SysMLv2 OWL rendering.
    No novel epistemic vocabulary in `rtm:` — adequacy and
    sufficiency are `gsn:Assumption` / `gsn:Justification` per the
    Hawkins–Habli ACP categorization.

    1. **The Assignment** —
       a. received four requirements from systems engineering,
          allocated to the ADCS subsystem;
       b. loaded the structural model (SysMLv2 RDF) into
          `<adcs:structural>`.
    2. **Symbolic Analysis** — SymPy derived 13 lemmas across 4 proof
       scripts (inertia tensor, stability margins, pointing budget,
       gravity-gradient bound, wheel-momentum bound). Each lemma is
       independently re-verifiable; the proof hash binds the script to
       a specific model hash.
    3. **Numerical Simulation** — scipy integrated quaternion + Euler
       dynamics under the PD controller. The step response revealed
       settling time ≈ 262s, exceeding the 120s spec for REQ-001 — the
       v1 engineering finding.
    4. **Evidence Binding** — emitted 4 proof artifacts + 3 simulation
       results into `<adcs:evidence>`, each with content / model / proof
       hashes and PROV-O provenance.
    5. **Attestation (GSN + EARL)** — emitted four well-formed
       attestations into `<adcs:attestations>`, each with an adequacy
       `gsn:Assumption`, a sufficiency `gsn:Justification`, an EARL
       outcome, and a `prov:qualifiedAssociation`. REQ-001 carries
       `earl:failed` rather than being silently omitted, keeping the
       audit graph complete.
    6. **Closure-Rule Validation + Audit (initial)** —
       a. ran the closure-rule suite (9 SHACL shapes + 1 runtime
          re-verification) against `rtm_graph` — all pass;
       b. ran the audit module — traceability is structurally intact
          (forward / backward / bidirectional all PASS); the coverage
          matrix shows REQ-001 = `covered+failed` (the expected v1
          state — engineering finding, *not* a trace gap).
    7. **Traceability Graph** — rendered the v1 RTM. Three green
       attestation nodes (REQ-002 / 3 / 4 = `earl:passed`), one red
       (REQ-001 = `earl:failed`). Design elements in purple to keep
       green / red reserved for the engineering verdict.
    8. **Design Iteration** —
       a. applied a SPARQL UPDATE to retune both gains (Kp: 1 → 4,
          Kd: 10 → 30) directly on the structural model;
       b. model hash invalidated — re-ran symbolic + numerical
          analysis, regenerated every proof and simulation hash from
          scratch (no stale evidence carries over);
       c. re-attested all four requirements as `earl:passed` against
          the v2 model.
    9. **Remote Compute & Distribution** —
       a. attached `ExecutionMetadata` (image digest, container ID,
          hostname, Python version) to v2 analysis activities,
          emulating analysis run on a remote `adcs-compute:latest`
          container;
       b. validated live persistence to Flexo MMS on the shared
          `try-layer1.starforge.app` sandbox — each named graph lands
          as its own branch under `adcs-demo/lifecycle`.
    10. **Fresh Audit (after remote compute)** — re-ran audit + closure
        rules against the v2 Dataset. Coverage matrix is now all
        `covered+passed`. A SPARQL query against the union view pulls
        per-activity *where & on what image was this produced* — the
        audit chain now extends all the way to the executor.

    ### What makes this different

    - **Evidence is not verification.** Only human attestation closes
      the loop — and attestation outcomes can be `earl:failed`,
      `earl:cantTell`, etc. The graph records the engineering verdict,
      not a binary pass/fail.
    - **Failures are first-class.** REQ-001 at v1 is `covered+failed`,
      not omitted. The audit trail is complete *because* the failure
      is recorded.
    - **Trace ≠ verdict.** Forward/backward traceability is a
      structural property of the graph. Coverage status is an
      engineering verdict per requirement. Act 6 calls these out as
      two orthogonal questions; the demo design relies on the
      distinction.
    - **Reproducibility is regression testing.** The pipeline that
      found the deficiency at v1 confirmed the fix didn't break
      anything else at v2.
    - **Model changes invalidate evidence.** Gain retune → model hash
      changes → every dependent hash changes. No stale proofs survive
      a structural-model edit.
    - **Where matters as much as what.** Provenance captures not just
      *what was computed* but *where and on what toolchain* (image
      digest + container ID + hostname), so a regulator can replay
      the analysis against the same pinned environment.
    - **Everything is in git, and everything composes.** Both model
      versions, both evidence sets, both attestation records — all
      text, all versioned. The same Dataset persists to disk
      (`--backend=local`), to Flexo MMS (`--backend=flexo`), or to
      Fuseki (`--backend=fuseki`) without changing a SPARQL query.

    ### Architecture

    | Layer | Technology | Purpose |
    |-------|-----------|---------|
    | Structural Model | SysMLv2 RDF/Turtle (aliased to OMG SysMLv2 OWL) | Requirements, design elements, satisfy links |
    | Model Changes | SPARQL UPDATE | Modify parameters, trigger hash invalidation |
    | Evidence Layer | PROV-O + EARL + OntoGSN + thin `rtm:` glue | Hash-bound evidence + GSN-structured attestation |
    | Process Model | P-PLAN | Declarative pipeline; one `p-plan:Step` per stage |
    | Closure Rules | SHACL + runtime re-verification | 10 well-formedness invariants enforced at Stage 6.5 |
    | Audit | Custom `traceability.audit` | Forward / backward / bidirectional with independence |
    | Storage | rdflib `Dataset` (named-graph quadstore) | Eight named graphs; Flexo/Fuseki-ready layout |
    | Symbolic Analysis | SymPy ProofScripts | Formal proofs with re-verifiable lemma chains |
    | Numerical Simulation | scipy solve_ivp | Time-domain ODE integration |
    | Compute Provenance | Local + Docker backends | `ExecutionMetadata` → PROV triples per activity |
    | Version Control | Git | Source of truth for all artifacts |
    """)
    return


@app.cell(hide_code=True)
def __():
    import marimo as mo
    return (mo,)


if __name__ == "__main__":
    app.run()