Upload 24 files

DETERMINISTIC_EXECUTION_GATE.md

@@ -0,0 +1,42 @@
+# Deterministic execution-integrity gate
+
+This revision prevents invalid hydraulic arrays from being presented as
+successful SWMR results.
+
+## Changes
+
+1. Dynamic-wave preflight rejects explicit `MAX_TRIALS <= 0` and
+   `HEAD_TOLERANCE <= 0`. Omitted values remain untouched so the engine default
+   applies. The uploaded INP is never edited.
+2. Post-run integrity classification records `valid`, `limited`, or `invalid`
+   using routing convergence and flow-routing continuity evidence.
+3. A run is invalid when every routing step fails, at least 5% of routing
+   steps fail, or the routing continuity error is at least 10%.
+4. Invalid hydraulic arrays remain available in the audit database, but node,
+   link, storage, control, capacity, spill, and depth-velocity conclusions are
+   set to `Not assessed - hydraulic routing solution invalid`.
+5. The executive summary and model identity disclose the execution gate.
+6. Calgary screening returns no pass/fail hydraulic results for an invalid run.
+7. Regression tests cover the Kincora zero-value failure and the corrected
+   usable-run metadata.
+
+## Kincora failure reproduced
+
+The failed input explicitly contained:
+
+```ini
+MAX_TRIALS       0
+HEAD_TOLERANCE   0
+```
+
+It produced 7,201/7,201 non-convergent steps and 100% routing continuity
+error. Removing those explicit zero overrides in a separate diagnostic copy
+restored 0 non-convergent steps and 0.022% routing continuity error. This
+revision rejects the failed configuration rather than silently changing it.
+
+## Validation
+
+- Python compilation passed for all packaged modules.
+- 30 deterministic unit checks passed.
+- Preflight integration requires the Space dependencies and isolated
+  OpenSWMM worker environment described in the existing README.

report_engine.py

@@ -647,13 +647,19 @@ def _critical_elements(node_df: pd.DataFrame, link_df: pd.DataFrame, units: Unit
 
 def _summary_findings(node_df: pd.DataFrame, link_df: pd.DataFrame, sub_df: pd.DataFrame, metadata: dict[str, Any], units: UnitContext, options: dict[str, str], criteria: ReportCriteria) -> list[str]:
     findings: list[str] = []
-    from screening_logic import continuity_disclosure
+    from screening_logic import continuity_disclosure, execution_integrity_assessment
+    integrity = execution_integrity_assessment(metadata)
+    if not integrity["results_usable"]:
+        findings.append("HYDRAULIC RESULTS INVALID: " + integrity["reason"])
     findings.extend(continuity_disclosure(
         metadata, review_pct=criteria.continuity_review,
         warning_pct=criteria.continuity_warning,
         has_pollutants=bool(metadata.get("has_pollutants"))))
     if False:  # legacy block replaced by deterministic continuity_disclosure
         pass
+    if not integrity["results_usable"]:
+        findings.append("All result-dependent hydraulic screening is Not assessed; raw arrays remain in the audit package only.")
+        return findings
     if node_df is not None and not node_df.empty:
         flooded_col = next((c for c in node_df.columns if c.startswith('Peak Flooding (')), None)
         flooded = int((pd.to_numeric(node_df[flooded_col], errors='coerce').fillna(0) > 0).sum()) if flooded_col else 0
@@ -726,6 +732,13 @@ def _event_summary(sub_df: pd.DataFrame, metadata: dict[str, Any], options: dict
 
 
 def _executive_summary(findings: list[str], critical_nodes: pd.DataFrame, critical_links: pd.DataFrame, units: UnitContext, criteria: ReportCriteria) -> list[str]:
+    invalid_line = next((x for x in findings if x.startswith("HYDRAULIC RESULTS INVALID:")), None)
+    if invalid_line:
+        return [
+            "The engine process completed, but the hydraulic routing solution failed the deterministic execution-integrity gate.",
+            invalid_line,
+            "Hydraulic arrays are retained for audit only. Capacity, surcharge, flooding, storage, control, spill-route and depth-velocity conclusions are Not assessed.",
+        ]
     lines = ["The simulation completed and the principal model results were screened using the project criteria listed in this report."]
     if critical_nodes.empty:
         lines.append(f"No junctions or dividers met the generic node screening criteria (depth ratio ≥ {criteria.node_depth_ratio:.2f}, rim clearance ≤ {criteria.minimum_freeboard:g} {units.length}, or flooding greater than zero). Storage facilities are assessed separately using their Calgary storage classification.")
@@ -1080,6 +1093,14 @@ def generate_report_package(
 ) -> dict[str, bytes | str]:
     """Generate editable Word report and ZIP package entirely in memory."""
     criteria = criteria or ReportCriteria()
+    simulation_metadata = dict(simulation_metadata or {})
+    from screening_logic import execution_integrity_assessment
+    integrity = execution_integrity_assessment(simulation_metadata)
+    simulation_metadata.update({
+        "execution_integrity_status": integrity["status"],
+        "results_usable": integrity["results_usable"],
+        "execution_integrity_reason": integrity["reason"],
+    })
     options = _inp_options(inp_sections)
     units = _unit_context(options.get("FLOW_UNITS", ""))
     sub_table = _subcatchment_model_table(inp_sections, sub_summary)
@@ -1117,7 +1138,6 @@ def generate_report_package(
                 storage_table.at[idx, "Status"] = "Review: control flow occurs but storage depth is zero"
     area_table = _area_classification(sub_table, units, criteria.area_classification)
     from screening_logic import effective_velocity_table, missing_information_register
-    simulation_metadata = dict(simulation_metadata or {})
     simulation_metadata["has_pollutants"] = bool(inp_sections.get("POLLUTANTS"))
     recon_links_df = None
     if reconciliation and reconciliation.get("links") is not None:
@@ -1148,6 +1168,18 @@ def generate_report_package(
     criteria_table = criteria_register(calgary)
     # Calgary-specific calculations use SI design rules. In US models, tables remain available but are flagged for review.
     minor_capacity = build_minor_system_capacity_table(link_table, units.system, units.flow, units.length, calgary)
+    if not integrity["results_usable"]:
+        invalid_label = "Not assessed - hydraulic routing solution invalid"
+        for df in (node_report, link_report, link_table, control_table, storage_table):
+            if df is not None and not df.empty:
+                df["Status"] = invalid_label
+        if control_recon is not None and not control_recon.empty:
+            control_recon["Check"] = "Indeterminate - hydraulic routing solution invalid"
+        if minor_capacity is not None and not minor_capacity.empty:
+            if "Status" in minor_capacity:
+                minor_capacity["Status"] = invalid_label
+            if "Assessment Basis" in minor_capacity:
+                minor_capacity["Assessment Basis"] = invalid_label
     # Checklist applicability is based on actual model element types, not on
     # whether the generic link table happens to contain outlet controls.
     _conduit_rows = inp_sections.get("CONDUITS", []) or []
@@ -1158,7 +1190,7 @@ def generate_report_package(
         metadata,
         criteria,
         has_model=bool(inp_sections),
-        has_results=not node_report.empty,
+        has_results=not node_report.empty and integrity["results_usable"],
         has_storage=not storage_table.empty,
         has_controls=not control_table.empty,
         has_overland=_has_overland,
@@ -1355,6 +1387,12 @@ def generate_report_package(
     overland_cols = ["Link ID", "From Node", "To Node", "Shape", f"Peak Flow ({units.flow})", f"Peak Depth ({units.length})", f"Peak Velocity ({units.velocity})", "Depth Ratio", "Status"]
     _add_df_table(doc, "Table 9 - Overland Flow Assessment", overland[[c for c in overland_cols if c in overland.columns]], landscape=True)
     overland_compliance = build_overland_compliance_table(overland, calgary, units.flow, units.length, units.velocity) if units.system == "SI" else pd.DataFrame()
+    if not integrity["results_usable"] and not overland_compliance.empty:
+        for col in ("Depth-Velocity Status", "Special Limit Status"):
+            if col in overland_compliance:
+                overland_compliance[col] = "Not assessed - hydraulic routing solution invalid"
+        if "Spill Active" in overland_compliance:
+            overland_compliance["Spill Active"] = "Not assessed"
     if not overland_compliance.empty:
         _add_df_table(doc, "Table 9A - Calgary Major-System Depth-Velocity Screening", overland_compliance, landscape=True, font_size=7.2)
 
@@ -1383,6 +1421,9 @@ def generate_report_package(
     doc.add_heading("4.4 Storage and Trap-Low Assessment", level=2)
     _add_narrative_block(doc, narrative_sections, "storage")
     storage_calgary = apply_storage_classification(storage_table, calgary, units.length) if not storage_table.empty else pd.DataFrame()
+    if not integrity["results_usable"] and not storage_calgary.empty:
+        if "Calgary Status" in storage_calgary:
+            storage_calgary["Calgary Status"] = "Not assessed - hydraulic routing solution invalid"
     if not criteria.suppress_empty_sections or not storage_calgary.empty:
         _add_df_table(doc, "Table 13 - Storage Unit Performance", storage_calgary, landscape=True, font_size=7.0)
     if criteria.suppress_empty_sections and control_table.empty and storage_table.empty:

screening_logic.py

@@ -22,6 +22,83 @@ from typing import Any, Mapping
 import pandas as pd
 
 
+# ---------------------------------------------------------------------------
+# Solver-option and execution-integrity gates
+# ---------------------------------------------------------------------------
+
+def validate_solver_options(options: Mapping[str, Any]) -> list[str]:
+    """Return deterministic blocking errors for unusable solver settings.
+
+    The uploaded engineering model remains immutable.  This function never
+    substitutes defaults or edits the INP; it only prevents a known-invalid
+    configuration from being presented as a completed hydraulic analysis.
+    """
+    opts = {str(k).upper(): str(v).strip() for k, v in options.items()}
+    if opts.get("FLOW_ROUTING", "").upper() != "DYNWAVE":
+        return []
+    errors: list[str] = []
+    for name, label in (("MAX_TRIALS", "hydraulic trials"),
+                        ("HEAD_TOLERANCE", "head tolerance")):
+        if name not in opts:
+            continue  # omitted means use the engine default
+        try:
+            value = float(opts[name])
+        except (TypeError, ValueError):
+            errors.append(f"{name} must be numeric for dynamic-wave routing.")
+            continue
+        if value <= 0:
+            errors.append(
+                f"{name} must be greater than zero for dynamic-wave routing; "
+                f"the uploaded value {opts[name]!r} disables a usable {label}."
+            )
+    return errors
+
+
+def execution_integrity_assessment(metadata: Mapping[str, Any]) -> dict[str, Any]:
+    """Classify whether hydraulic results can support screening conclusions."""
+    def number(key: str, default: float = 0.0) -> float:
+        try:
+            return float(metadata.get(key, default))
+        except (TypeError, ValueError):
+            return default
+
+    steps = int(number("routing_steps"))
+    failed = int(number("not_converged_steps"))
+    pct_failed = number("pct_not_converged")
+    flow_error = abs(number("flow_error"))
+    runoff_error = abs(number("runoff_error"))
+
+    invalid_reasons: list[str] = []
+    if steps > 0 and failed >= steps:
+        invalid_reasons.append("every routing step failed to converge")
+    elif pct_failed >= 5.0:
+        invalid_reasons.append(f"{pct_failed:.3f}% of routing steps failed to converge")
+    if flow_error >= 10.0:
+        invalid_reasons.append(f"flow-routing continuity error is {flow_error:.3f}%")
+
+    if invalid_reasons:
+        return {
+            "status": "invalid",
+            "results_usable": False,
+            "hydraulic_conclusions_allowed": False,
+            "reason": "; ".join(invalid_reasons) + ".",
+        }
+
+    limitations: list[str] = []
+    if failed > 0:
+        limitations.append(f"{failed} routing step(s) did not converge")
+    if flow_error > 1.0:
+        limitations.append(f"flow-routing continuity error is {flow_error:.3f}%")
+    if runoff_error > 1.0:
+        limitations.append(f"runoff continuity error is {runoff_error:.3f}%")
+    return {
+        "status": "limited" if limitations else "valid",
+        "results_usable": True,
+        "hydraulic_conclusions_allowed": True,
+        "reason": "; ".join(limitations) + ("." if limitations else "Execution-integrity checks passed."),
+    }
+
+
 # ---------------------------------------------------------------------------
 # Velocity classification and evidence precedence
 # ---------------------------------------------------------------------------

test_regression.py

@@ -41,8 +41,30 @@ def check(name: str, cond: bool, detail: str = "") -> None:
 def unit_tests() -> None:
     print("\n== Unit tests: screening_logic ==")
     from screening_logic import (classify_velocity, continuity_disclosure,
+                                 execution_integrity_assessment,
                                  effective_velocity_table,
-                                 missing_information_register)
+                                 missing_information_register,
+                                 validate_solver_options)
+
+    solver_errors = validate_solver_options({
+        "FLOW_ROUTING": "DYNWAVE", "MAX_TRIALS": "0", "HEAD_TOLERANCE": "0"})
+    check("zero MAX_TRIALS rejected", any("MAX_TRIALS" in x for x in solver_errors))
+    check("zero HEAD_TOLERANCE rejected", any("HEAD_TOLERANCE" in x for x in solver_errors))
+    check("omitted dynamic-wave overrides use engine defaults",
+          validate_solver_options({"FLOW_ROUTING": "DYNWAVE"}) == [])
+
+    invalid = execution_integrity_assessment({
+        "routing_steps": 7201, "not_converged_steps": 7201,
+        "pct_not_converged": 100.0, "flow_error": 100.0,
+        "runoff_error": -1.933})
+    check("100% nonconvergence invalidates results", invalid["status"] == "invalid")
+    check("invalid run blocks hydraulic conclusions",
+          not invalid["results_usable"] and not invalid["hydraulic_conclusions_allowed"])
+    valid = execution_integrity_assessment({
+        "routing_steps": 12641, "not_converged_steps": 0,
+        "pct_not_converged": 0.0, "flow_error": 0.022,
+        "runoff_error": -1.933})
+    check("corrected Kincora run remains usable", valid["results_usable"])
 
     check("advisory classification (3.5)", "Advisory" in classify_velocity(3.5))
     check("critical classification (4.5)", "Critical" in classify_velocity(4.5))

tools.py

@@ -33,6 +33,7 @@ from results_db import ResultDatabase
 from sessions import STORE
 from sql_agent import SafeSQLAgent
 from swmm_core import run_swmm
+from screening_logic import execution_integrity_assessment, validate_solver_options
 
 MAX_ROWS = 60
 MAX_TS_POINTS = 200
@@ -54,160 +55,47 @@ def _require_results(session) -> None:
         raise ValueError(f"Session '{session.id}' has no simulation results yet. Call run_simulation first.")
 
 
+def _options_map(sections: dict[str, list[list[str]]]) -> dict[str, str]:
+    return {str(r[0]).upper(): str(r[1]) for r in sections.get("OPTIONS", []) if len(r) >= 2}
+
+
 # ---------------------------------------------------------------------------
 # Model lifecycle
 # ---------------------------------------------------------------------------
 
-MAX_MODEL_BYTES = 20 * 1024 * 1024
-
-
-def _decode_model_payload(
-    inp_content: str,
-    filename: str = "model.inp",
-    encoding: str = "auto",
-) -> tuple[str, str]:
-    """Normalize raw text, base64, or a JSON-wrapped SWMM model payload."""
-    if not isinstance(inp_content, str) or not inp_content.strip():
-        raise ValueError("inp_content is required and must be a non-empty string.")
-
-    value = inp_content.strip()
-    resolved_filename = str(filename or "model.inp").strip() or "model.inp"
-    selected_encoding = str(encoding or "auto").strip().lower()
-
-    # Accept a complete JSON object passed as the inp_content value:
-    # {"filename": "...", "inp_content": "...", "encoding": "base64|raw|auto"}
-    if value.startswith("{"):
-        try:
-            payload = json.loads(value)
-        except json.JSONDecodeError:
-            payload = None
-
-        if isinstance(payload, dict):
-            if "inp_content" not in payload:
-                raise ValueError("JSON upload payload must contain an 'inp_content' field.")
-            value = str(payload.get("inp_content") or "").strip()
-            resolved_filename = (
-                str(payload.get("filename") or "").strip()
-                or resolved_filename
-            )
-            selected_encoding = str(
-                payload.get("encoding") or selected_encoding
-            ).strip().lower()
-
-    if not value:
-        raise ValueError("The upload payload contains no SWMM model content.")
-
-    if selected_encoding in {"raw", "text"}:
-        model_text = value
-
-    elif selected_encoding == "base64":
-        try:
-            decoded = base64.b64decode(value, validate=True)
-        except (binascii.Error, ValueError) as exc:
-            raise ValueError("Invalid base64 SWMM model content.") from exc
-        try:
-            model_text = decoded.decode("utf-8-sig")
-        except UnicodeDecodeError:
-            model_text = decoded.decode("cp1252")
-
-    elif selected_encoding == "auto":
-        # Raw INP text normally exposes a section header near the beginning.
-        if "[" in value[:2000]:
-            model_text = value
-        else:
-            try:
-                decoded = base64.b64decode(value, validate=True)
-                try:
-                    candidate = decoded.decode("utf-8-sig")
-                except UnicodeDecodeError:
-                    candidate = decoded.decode("cp1252")
-            except (binascii.Error, ValueError, UnicodeDecodeError):
-                candidate = ""
-
-            model_text = candidate if "[" in candidate[:2000] else value
-
-    else:
-        raise ValueError("encoding must be one of: auto, raw, text, or base64.")
-
-    encoded_size = len(model_text.encode("utf-8"))
-    if encoded_size > MAX_MODEL_BYTES:
-        raise ValueError(
-            f"SWMM model exceeds the {MAX_MODEL_BYTES // (1024 * 1024)} MB upload limit."
-        )
-
-    upper = model_text.upper()
-    if "[OPTIONS]" not in upper and "[JUNCTIONS]" not in upper:
-        raise ValueError(
-            "Content does not look like a SWMM .inp file "
-            "(no [OPTIONS] or [JUNCTIONS] section)."
-        )
-
-    safe_name = Path(resolved_filename).name or "model.inp"
-    if not safe_name.lower().endswith(".inp"):
-        safe_name = f"{Path(safe_name).stem}.inp"
-
-    return model_text, safe_name
-
-
-def upload_model(
-    inp_content: str,
-    filename: str = "model.inp",
-    encoding: str = "auto",
-) -> dict:
-    """Upload an EPA SWMM model as raw text, base64, or a JSON-wrapped payload.
-
-    JSON payload format:
-    {"filename": "model.inp", "inp_content": "...", "encoding": "auto|raw|base64"}
+def upload_model(inp_content: str, filename: str = "model.inp") -> dict:
+    """Upload an EPA SWMM .inp model (raw text or base64) and create a session.
 
     Returns a session_id used by every other tool, plus element counts.
     """
-    model_text, safe_name = _decode_model_payload(
-        inp_content=inp_content,
-        filename=filename,
-        encoding=encoding,
-    )
-
+    text = inp_content
+    if "[" not in inp_content[:2000]:  # likely base64
+        try:
+            text = base64.b64decode(inp_content, validate=True).decode("utf-8", errors="replace")
+        except (binascii.Error, ValueError):
+            pass
+    if "[OPTIONS]" not in text.upper() and "[JUNCTIONS]" not in text.upper():
+        raise ValueError("Content does not look like a SWMM .inp file (no [OPTIONS]/[JUNCTIONS] section).")
     session = STORE.create()
-    try:
-        inp_path = session.workdir / safe_name
-        inp_path.write_text(model_text, encoding="utf-8", newline="\n")
-
-        sections = mp.parse_inp_sections(str(inp_path))
-        if not sections:
-            raise ValueError("The uploaded file could not be parsed into SWMM sections.")
-
-        session.data.update({
-            "filename": safe_name,
-            "inp_path": str(inp_path),
-            "sections": sections,
-            "upload_encoding": encoding,
-            "model_size_bytes": inp_path.stat().st_size,
-        })
-
-        counts = {
-            name: len(rows)
-            for name, rows in sections.items()
-            if name in (
-                "JUNCTIONS", "OUTFALLS", "STORAGE", "CONDUITS", "PUMPS",
-                "WEIRS", "ORIFICES", "OUTLETS", "SUBCATCHMENTS",
-                "RAINGAGES", "TIMESERIES",
-            )
-        }
-        gages = [row[0] for row in sections.get("RAINGAGES", []) if row]
-
-        return {
-            "session_id": session.id,
-            "filename": safe_name,
-            "model_size_bytes": inp_path.stat().st_size,
-            "element_counts": counts,
-            "rain_gages": gages,
-            "design_event_inference": infer_design_event(gages) if gages else None,
-            "next_step": "Call run_simulation with this session_id.",
-        }
-
-    except Exception:
-        STORE.drop(session.id)
-        raise
+    safe_name = Path(filename).name or "model.inp"
+    if not safe_name.lower().endswith(".inp"):
+        safe_name += ".inp"
+    inp_path = session.workdir / safe_name
+    inp_path.write_text(text, encoding="utf-8")
+    sections = mp.parse_inp_sections(str(inp_path))
+    session.data.update({"filename": safe_name, "inp_path": str(inp_path), "sections": sections})
+    counts = {name: len(rows) for name, rows in sections.items()
+              if name in ("JUNCTIONS", "OUTFALLS", "STORAGE", "CONDUITS", "PUMPS", "WEIRS",
+                          "ORIFICES", "OUTLETS", "SUBCATCHMENTS", "RAINGAGES", "TIMESERIES")}
+    gages = [row[0] for row in sections.get("RAINGAGES", []) if row]
+    return {
+        "session_id": session.id,
+        "filename": safe_name,
+        "element_counts": counts,
+        "rain_gages": gages,
+        "design_event_inference": infer_design_event(gages) if gages else None,
+        "next_step": "Call run_simulation with this session_id.",
+    }
 
 
 def run_simulation(session_id: str) -> dict:
@@ -219,6 +107,13 @@ def run_simulation(session_id: str) -> dict:
     inp_path = session.data.get("inp_path")
     if not inp_path:
         raise ValueError("Session has no uploaded model.")
+    sections = session.data["sections"]
+    option_errors = validate_solver_options(_options_map(sections))
+    if option_errors:
+        session.data["input_validation_errors"] = option_errors
+        raise ValueError(
+            "Input validation failed; simulation was not run. " + " ".join(option_errors)
+        )
     results = run_swmm(inp_path)
     md = results["metadata"]
     import hashlib, datetime
@@ -226,12 +121,24 @@ def run_simulation(session_id: str) -> dict:
     run_id = datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
     md["model_sha256"] = sha256
     md["run_id"] = run_id
-    sections = session.data["sections"]
+    integrity = execution_integrity_assessment(md)
+    md.update({
+        "execution_integrity_status": integrity["status"],
+        "results_usable": integrity["results_usable"],
+        "hydraulic_conclusions_allowed": integrity["hydraulic_conclusions_allowed"],
+        "execution_integrity_reason": integrity["reason"],
+    })
     node_df = mp.build_node_summary(results["node_ts"], mp.parse_node_types(sections), 0.001, 0.9)
     link_df = mp.build_link_summary(results["link_ts"], mp.parse_link_topology(sections),
                                     mp.parse_conduit_geometry(sections), 0.9, 3.0)
     sub_df = mp.build_sub_summary(results["sub_ts"], mp.parse_subcatchment_attrs(sections),
                                   results.get("times"), md.get("flow_units", "CMS"))
+    if not integrity["results_usable"]:
+        invalid_label = "Not assessed - hydraulic routing solution invalid"
+        if "Status" in node_df:
+            node_df["Status"] = invalid_label
+        if "Status" in link_df:
+            link_df["Status"] = invalid_label
     db = ResultDatabase(str(session.workdir / "results.sqlite"))
     db.load(node_df, link_df, sub_df, inp_path=inp_path, results=results)
 
@@ -253,7 +160,9 @@ def run_simulation(session_id: str) -> dict:
     flooded = int((pd.to_numeric(node_df[flooded_col], errors="coerce").fillna(0) > 0.001).sum()) if flooded_col else 0
     return {
         "session_id": session.id,
-        "simulation": "completed",
+        "simulation": "completed" if integrity["results_usable"] else "completed_invalid",
+        "execution_integrity": integrity,
+        "results_usable": integrity["results_usable"],
         "model_sha256": sha256,
         "run_id": run_id,
         "flow_units": md.get("flow_units"),
@@ -262,7 +171,9 @@ def run_simulation(session_id: str) -> dict:
         "warnings": (results.get("warnings") or md.get("warnings") or [])[:10],
         "flooded_nodes": flooded,
         "rpt_reconciliation": recon,
-        "note": "Values are model results, not engineering determinations.",
+        "note": ("Values are model results, not engineering determinations."
+                 if integrity["results_usable"] else
+                 "Hydraulic arrays are retained for audit only and must not be used for screening conclusions."),
     }
 
 
@@ -391,6 +302,15 @@ def calgary_screening(session_id: str) -> dict:
     """
     session = STORE.get(session_id)
     _require_results(session)
+    integrity = execution_integrity_assessment(session.data["results"].get("metadata", {}))
+    if not integrity["results_usable"]:
+        return {
+            "velocity_screen_flagged": _df_records(pd.DataFrame(), 30),
+            "storage_classification": _df_records(pd.DataFrame(), 30),
+            "criteria_register": _df_records(criteria_register(CalgaryCriteria()), 40),
+            "status": "Not assessed - hydraulic routing solution invalid",
+            "execution_integrity": integrity,
+        }
     crit = CalgaryCriteria()
     link_df = session.data["link_df"]
     node_df = session.data["node_df"]
@@ -419,7 +339,8 @@ def preliminary_design_review(session_id: str) -> dict:
         inp_text=inp_text, node_summary=session.data["node_df"],
         link_summary=session.data["link_df"], sub_summary=session.data["sub_df"],
         metadata=session.data["results"]["metadata"],
-        simulation_completed=True, output_results_available=True)
+        simulation_completed=True,
+        output_results_available=bool(session.data["results"]["metadata"].get("results_usable", True)))
     recon_findings = rr.reconciliation_findings(
         session.data.get("recon_links"), session.data.get("recon_nodes"),
         session.data.get("recon_continuity"))
@@ -623,7 +544,9 @@ def generate_report(session_id: str, project_name: str, client: str = "",
                         "session_id": session.id,
                         "run_id": session.data["results"]["metadata"].get("run_id", "—"),
                         "engine": "EPA SWMM / OpenSWMM 6 (crash-isolated worker, engine Rev 23.2)",
-                        "status": "Completed"})
+                        "status": ("Completed - results usable" if
+                                   session.data["results"]["metadata"].get("results_usable", True)
+                                   else "Completed - hydraulic results invalid")})
     outputs = session.workdir / "outputs"
     outputs.mkdir(exist_ok=True)
     files = {}
@@ -652,4 +575,4 @@ TOOL_REGISTRY: dict[str, Callable[..., dict]] = {
         calgary_screening, preliminary_design_review, get_reconciliation,
         run_scenario, attach_figure, set_report_details, generate_report,
     ]
-}
+}