feat(optimization): persist and revalidate field envelope evidence

.github/skills/neqsim-continuous-task-improvement/SKILL.md

@@ -198,6 +198,25 @@ writes to a control system and every proposal carries `requires_approval`.
   candidates. It keeps only proposals that also satisfy every hard constraint with margin; every rejection is
   listed in `guard.json`. Also record a positive integer `evaluation_count` and finite non-negative
   `runtime_seconds`; the ledger retains the evidence block for persistent incumbent and performance provenance.
+  Field studies can set `production.require_envelope_evidence: true` and attach the Java
+  `getAcceptedPointEnvelopeEvidence(...)` map as `accepted_point_envelope_evidence`. The guard requires exact
+  cycle/optimizer identity, action-id/setpoint binding, complete uniquely addressed slices, safe recovery of every
+  sampled point, and reconciled point/evaluation/runtime totals before the existing ledger persists it. A supplied
+  block is always validated even when the setting is false; never copy envelope evidence into a later cycle.
+  Map production/injection route line-ups, controlled reservoir-lifecycle boundaries, mapped rotating-equipment
+  power, common-shaft speed/power, driver limits, shared host-power limits, and other operating-limit files to
+  `guard` with dependencies on the model update, optimizer, gates, and constraints. Keep route selectors as process
+  actions, reservoir pressure as a model input owned by `WellFlow.ReservoirLayer`, and mapped equipment power as
+  solved evidence read from the canonical `ProcessModel.getPower(...)` path. On a detected relevant revision
+  the append-only ledger marks old validation evidence non-current
+  and supersedes unfinished advice during the evidence stage, before model/optimizer/guard work can fail. A failed
+  guard still blocks publication of new advice, but cannot leave the old item marked current. Retry and resume
+  preserve idempotent invalidation and the current cycle's already-persisted replacement; dry runs do not mutate the
+  ledger. Custom production plans must keep `evidence` before model update, optimization, and `guard` to retain this
+  failure-path guarantee. Only a newly guarded point becomes a separate
+  `proposed` item with `revalidation_of`; if no safe replacement exists, expose none and never retain stale advice.
+  When a later revision is safe, resume lineage from the most recent superseded item rather than silently starting
+  a disconnected incumbent.
 - **Outcome**: ledger items set to `implemented` with `objective_kpi` and `baseline_value` are compared with
   the realised gain each cycle (`outcome_confirmed` / `outcome_miss`); the engineer decides on `verified`.
 - **Boundary conditions**: read the measured process boundary (here gas export and injection pressure and

devtools/neqsim_continuous/cycle.py

@@ -3,7 +3,8 @@
 A cycle runs the planned stages in order, writes an immutable folder
 ``continuous/cycles/<cycle_id>/`` and a ``cycle.json`` manifest, and never raises for a
 stage failure: the failure is recorded, dependants are skipped and the cycle is marked
-``degraded``. An interrupted cycle resumes at its first incomplete stage.
+``degraded``. An interrupted cycle reuses completed work and revalidates production
+gates, constraints and proposals before publishing advice.
 """
 
 import csv
@@ -24,8 +25,12 @@
 
 SCHEMA_VERSION = "1.0"
 NEEDS = {"drift": ["kpis"], "diff": ["kpis"], "goal": ["kpis"], "notify": ["digest"],
-         "gates": ["kpis"], "constraints": ["kpis"], "guard": ["gates", "constraints"], "outcome": ["kpis"]}
+         "gates": ["kpis"], "constraints": ["kpis"], "guard": ["gates", "constraints"],
+         "ledger": ["guard"], "outcome": ["kpis"]}
 LOCK_HOURS = 6.0
+# These cheap checks mutate in-memory blocks/proposals, not model or optimizer state.
+# Restoring their status alone would lose the filtering performed before interruption.
+REVALIDATE_ON_RESUME = ("gates", "constraints", "guard")
 
 
 def _utc(value=None):
@@ -58,6 +63,7 @@ def __init__(self, task_dir, plan, goal, baseline, now, mode, cycle_id, cycle_di
         self.solve, self.stop_state, self.digest_text = {}, None, ""
         self.notifications, self.agent_run, self.previous_kpis = [], {}, {}
         self.guard_blocks, self.sections = [], []
+        self.guarded_proposals = set()
         self.user_levers = {}
         self.evidence_analysis = evidence_analysis
 
@@ -229,14 +235,21 @@ def run_cycle(task_dir, mode="monitor", now=None, stages=None, dry_run=False, no
             spec = dict(spec, name=name)
             started = time.time()
             proposal_start = len(ctx.proposals)
-            if name in done:
+            if name in done and name not in REVALIDATE_ON_RESUME:
                 saved = done[name]
                 result = StageResult(**{k: v for k, v in saved.items() if k in (
                     "name", "status", "outputs", "kpis", "triggers", "message")})
                 if name == "refresh":
                     _builtin_stages.restore_refresh(ctx, result)
                 elif name == "evidence":
                     _builtin_stages.restore_evidence(ctx)
+                elif name == "inputs":
+                    try:
+                        _user_input_stage.restore_inputs(ctx)
+                    except Exception as error:
+                        ctx.guard_blocks.append("user_input_resume_state_invalid")
+                        result = StageResult(name, "fail", message="{}: {}".format(
+                            type(error).__name__, error))
                 if name.startswith("script:") or name == "kpis":
                     ctx.kpis.update(result.kpis)
                 if saved.get("solve"):

devtools/neqsim_continuous/ledger.py

@@ -62,6 +62,10 @@ def current(self):
                     item["status"] = event["to"]
                 elif kind == "value_updated":
                     item["value"] = event["value"]
+                elif kind == "evidence_invalidated":
+                    item["evidence_current"] = False
+                    item["evidence_invalidated_cycle"] = event.get("cycle")
+                    item["evidence_invalidated_by"] = list(event.get("changes") or [])
                 item["history"].append({k: v for k, v in event.items() if k != "id"})
         return items
 
@@ -102,6 +106,16 @@ def update_value(self, key, value, cycle=""):
             raise LedgerError("Unknown opportunity '{}'".format(key))
         return self._append({"event": "value_updated", "id": key, "value": value, "cycle": cycle})
 
+    def invalidate_evidence(self, key, changes, cycle=""):
+        """Mark persisted validation evidence stale without rewriting its history."""
+        if key not in self.current():
+            raise LedgerError("Unknown opportunity '{}'".format(key))
+        compact = [{name: row.get(name) for name in
+                    ("path", "change", "before_sha256", "after_sha256", "impact_rules")
+                    if row.get(name) is not None} for row in changes or []]
+        return self._append({"event": "evidence_invalidated", "id": key,
+                             "cycle": cycle, "changes": compact})
+
     def merge(self, other_path):
         """Union this ledger with another copy; returns the number of new events."""
         mine = self.events()

devtools/neqsim_continuous/living.py

@@ -72,16 +72,32 @@
 gates: []
 #  - {name: sep_T_residual, kpi: max_sep_T_residual_K, abs_max: 3.0}      # no advice if the model is off
 #  - {name: mass_balance, kpi: mass_balance_pct, abs_max: 0.05}
-production: {min_gain: 0.0, outcome_tolerance: 0.5}
+production:
+  min_gain: 0.0
+  outcome_tolerance: 0.5
+  require_envelope_evidence: false
+  evidence_sensitive_stages: ["script:model_update", "script:optimize", gates, constraints, guard]
 # Each proposal must include optimizer_evidence for this cycle: converged and finite candidate,
 # complete constraint evidence, complete state restoration, accepted-point replay, and all actions applied.
+# Set require_envelope_evidence true for field studies that must persist accepted-point
+# multi-slice/bottleneck evidence bound exactly to the proposed action identifiers and values.
 drift:
   signals: []
   settings: {lambda: 0.2, L: 3.5, k: 0.5, h: 6.0, warmup: 30}
 triggers:
   kpi_step: {}
   criteria: {}
-evidence: {rules: [], dependencies: {}}
+evidence:
+  rules: []
+# - name: field-operating-definition
+#   matches: [study_config.yaml, continuous/goal.yaml]
+#   stages: [guard]
+#   conclusions: [validated field operating point and envelope]
+  dependencies:
+    guard: ["script:optimize", gates, constraints]
+    "script:optimize": ["script:model_update"]
+    gates: ["script:model_update"]
+    constraints: ["script:model_update"]
 stages: [sense, evidence, inputs, refresh, "script:model_update", "script:optimize", kpis, gates, constraints, guard,
          drift, goal, diff, outcome, ledger, digest, notify, agent]
 notify:

devtools/neqsim_continuous/living_report.py

@@ -278,11 +278,18 @@ def build(task_dir):
         for key, item in sorted(items.items()):
             decided = [h for h in item.get("history", []) if h.get("event") == "status_changed"]
             value = item.get("value")
+            evidence = "-"
+            if item.get("evidence_current") is True:
+                evidence = "current"
+            elif item.get("evidence_current") is False:
+                evidence = "stale since {}".format(item.get("evidence_invalidated_cycle") or "unknown cycle")
             rows.append((key, item.get("status"), item.get("category"), item.get("title"),
                          value.get("value") if isinstance(value, dict) else value,
+                         evidence,
                          "{} {}".format(decided[-1].get("by"), str(decided[-1].get("at", ""))[:10])
                          if decided else "-"))
-        out += _table(["Id", "Status", "Category", "Title", "Latest value", "Last decision"], rows)
+        out += _table(["Id", "Status", "Category", "Title", "Latest value", "Evidence",
+                       "Last decision"], rows)
         waiting = [k for k, v in items.items() if v.get("status") == "proposed"]
         if waiting:
             out += ["", "Waiting for a decision: " + ", ".join(sorted(waiting)) +

devtools/neqsim_continuous/production.py

@@ -16,6 +16,7 @@
 The optimiser itself stays a task-local ``script:`` stage that returns ``proposals``.
 """
 
+import json
 import math
 import os
 
@@ -193,15 +194,150 @@ def _optimizer_evidence_reasons(ctx, proposal):
     return reasons
 
 
+def _positive_integer(value):
+    """Return true only for a positive, finite integer value."""
+    return _number(value) and value >= 1 and float(value) == int(value)
+
+
+def _non_negative_integer(value):
+    """Return true only for a non-negative, finite integer value."""
+    return _number(value) and value >= 0 and float(value) == int(value)
+
+
+def _same_finite_vector(first, second):
+    """Return true for equal-length, exactly equal finite numeric vectors."""
+    if not isinstance(first, list) or not isinstance(second, list) or len(first) != len(second):
+        return False
+    return all(_number(left) and _number(right) and float(left) == float(right)
+               for left, right in zip(first, second))
+
+
+def _envelope_evidence_reasons(ctx, proposal, required=False):
+    """Validate a persisted accepted-point envelope without trusting its Java origin."""
+    if "accepted_point_envelope_evidence" not in proposal:
+        return ["missing_accepted_point_envelope_evidence"] if required else []
+    evidence = proposal.get("accepted_point_envelope_evidence")
+    if not isinstance(evidence, dict):
+        return ["invalid_accepted_point_envelope_evidence"]
+
+    reasons = []
+    if evidence.get("schema_version") != 1:
+        reasons.append("unsupported_envelope_schema")
+    if evidence.get("cycle_id") != ctx.cycle_id:
+        reasons.append("stale_envelope_evidence")
+    if evidence.get("optimizer_evidence") != proposal.get("optimizer_evidence"):
+        reasons.append("envelope_optimizer_evidence_mismatch")
+
+    action_ids = evidence.get("accepted_action_ids")
+    accepted = evidence.get("accepted_candidate_values")
+    setpoints = proposal.get("setpoints")
+    valid_actions = (isinstance(action_ids, list) and action_ids
+                     and all(isinstance(item, str) and item for item in action_ids)
+                     and len(set(action_ids)) == len(action_ids))
+    accepted_ids = action_ids if valid_actions else []
+    if not valid_actions or not isinstance(accepted, list) or len(accepted) != len(accepted_ids):
+        reasons.append("invalid_accepted_action_vector")
+    elif (not isinstance(setpoints, dict) or set(setpoints) != set(accepted_ids)
+          or not _same_finite_vector([setpoints.get(action_id) for action_id in accepted_ids], accepted)):
+        reasons.append("envelope_setpoints_mismatch")
+
+    slices = evidence.get("slices")
+    slice_count = evidence.get("slice_count")
+    if not _positive_integer(slice_count) or not isinstance(slices, list) or len(slices) != int(slice_count or 0):
+        reasons.append("invalid_envelope_slice_count")
+        return reasons
+
+    requested_total = 0
+    completed_total = 0
+    evaluation_total = 0
+    runtime_total = 0.0
+    axis_signatures = set()
+    for index, row in enumerate(slices):
+        prefix = "envelope_slice_{}".format(index)
+        if not isinstance(row, dict):
+            reasons.append(prefix + "_invalid")
+            continue
+        if (row.get("study_id") != evidence.get("study_id")
+                or row.get("candidate_evaluator_id") != evidence.get("candidate_evaluator_id")):
+            reasons.append(prefix + "_identity_mismatch")
+        if row.get("complete") is not True or row.get("outcome") != "COMPLETE":
+            reasons.append(prefix + "_incomplete")
+        if not _same_finite_vector(row.get("anchor_values"), accepted):
+            reasons.append(prefix + "_anchor_mismatch")
+
+        axes = row.get("axis_action_ids")
+        if (not isinstance(axes, list) or not axes
+                or any(not isinstance(axis, str) or axis not in accepted_ids for axis in axes)
+                or len(set(axes)) != len(axes)):
+            reasons.append(prefix + "_invalid_axes")
+        else:
+            signature = tuple(axes)
+            if signature in axis_signatures:
+                reasons.append("duplicate_envelope_axes")
+            axis_signatures.add(signature)
+
+        requested = row.get("requested_point_count")
+        completed = row.get("completed_point_count")
+        evaluations = row.get("evaluation_count")
+        runtime = row.get("runtime_seconds")
+        points = row.get("points")
+        if (not _positive_integer(requested) or not _positive_integer(completed)
+                or int(requested) != int(completed)
+                or not _positive_integer(evaluations)
+                or not _number(runtime) or runtime < 0
+                or not isinstance(points, list) or len(points) != int(completed or 0)):
+            reasons.append(prefix + "_invalid_counts")
+        else:
+            requested_total += int(requested)
+            completed_total += int(completed)
+            evaluation_total += int(evaluations)
+            runtime_total += float(runtime)
+        if isinstance(points, list):
+            for point in points:
+                if (not isinstance(point, dict) or point.get("baseline_restored") is not True
+                        or point.get("baseline_simulation_converged") is not True):
+                    reasons.append(prefix + "_point_recovery_incomplete")
+                    break
+                if (not isinstance(point.get("candidate_values"), list)
+                        or len(point["candidate_values"]) != len(accepted_ids)
+                        or any(not _number(value) for value in point["candidate_values"])):
+                    reasons.append(prefix + "_invalid_point_candidate")
+                    break
+
+    if (evidence.get("requested_point_count") != requested_total
+            or evidence.get("completed_point_count") != completed_total
+            or evidence.get("envelope_evaluation_count") != evaluation_total):
+        reasons.append("envelope_aggregate_count_mismatch")
+    envelope_runtime = evidence.get("envelope_runtime_seconds")
+    if (not _number(envelope_runtime) or envelope_runtime < 0
+            or not math.isclose(float(envelope_runtime), runtime_total, rel_tol=1.0e-12, abs_tol=1.0e-12)):
+        reasons.append("envelope_runtime_mismatch")
+    optimizer = proposal.get("optimizer_evidence")
+    optimizer_count = optimizer.get("evaluation_count") if isinstance(optimizer, dict) else None
+    total_count = evidence.get("total_evaluation_count")
+    if (not _non_negative_integer(total_count) or not _positive_integer(optimizer_count)
+            or int(total_count) != int(optimizer_count) + evaluation_total):
+        reasons.append("envelope_total_evaluation_count_mismatch")
+    if evidence.get("advisory") is not True:
+        reasons.append("envelope_not_advisory")
+    return reasons
+
+
 def _non_finite_fields(values):
     """Return sorted keys whose proposed numeric values are absent or non-finite."""
     if not isinstance(values, dict):
         return ["<mapping>"]
     return sorted(str(key) for key, value in values.items() if not _number(value))
 
 
+def _proposal_snapshot(proposal):
+    """Bind publication to exact JSON content, including evidence boolean types."""
+    return json.dumps(proposal, sort_keys=True, separators=(",", ":"))
+
+
 def guard(ctx, spec):
     """Keep only proposals that are safe to show; record why the others were withheld."""
+    ctx.guarded_proposals = set()
     settings = ctx.plan.get("production", {}) or {}
     configured_min_gain = settings.get("min_gain", 0.0)
     invalid_min_gain = not _number(configured_min_gain)
@@ -234,6 +370,8 @@ def guard(ctx, spec):
         if not isinstance(proposal.get("objective_kpi"), str) or not proposal.get("objective_kpi"):
             reasons.append("missing_objective_kpi")
         reasons.extend(_optimizer_evidence_reasons(ctx, proposal))
+        reasons.extend(_envelope_evidence_reasons(
+            ctx, proposal, required=settings.get("require_envelope_evidence") is True))
         for constraint in hard:
             name = constraint.get("name") or constraint["kpi"]
             room = slack(constraint, predicted.get(constraint["kpi"]))
@@ -246,6 +384,7 @@ def guard(ctx, spec):
         else:
             accepted.append(dict(proposal, requires_approval=True, guard="passed"))
     ctx.proposals = accepted
+    ctx.guarded_proposals = {_proposal_snapshot(proposal) for proposal in accepted}
     write_json(os.path.join(ctx.cycle_dir, "guard.json"), {"accepted": [p.get("title") for p in accepted],
                                                             "withheld": withheld, "blocks": list(ctx.guard_blocks)})
     _section(ctx, "Proposals withheld", ["{} ({})".format(w["title"], ", ".join(w["reasons"])) for w in withheld])