app.py

import gradio as gr
import json
import uuid
from datetime import datetime, timezone

APP_TITLE = "Recursive Engine Observatory"

def now():
    return datetime.now(timezone.utc).isoformat()

def make_event(iteration, prior_state, trigger, observation, interpretation,
               action, artifact, verification, new_information,
               subsequent_influence=None, correction=None, next_iteration_input=None,
               prior_event_id=None):
    return {
        "eventId": f"evt_{uuid.uuid4().hex[:10]}",
        "eventType": "recursive.cycle.completed",
        "actorType": "billy",
        "iteration": iteration,
        "createdAt": now(),
        "priorEventId": prior_event_id,
        "recursive": {
            "priorState": prior_state,
            "trigger": trigger,
            "observation": observation,
            "interpretation": interpretation,
            "action": action,
            "artifact": artifact,
            "verification": verification,
            "newInformation": new_information,
            "subsequentInfluence": subsequent_influence,
            "correction": correction,
            "nextIterationInput": next_iteration_input,
        },
    }

def iteration_one(task, evidence, environment):
    task = task.strip() or "Determine whether Feature X should be inspected next."
    evidence = evidence.strip() or "The feature has recent activity, but the available evidence is incomplete."
    environment = environment.strip() or "The inspection reveals a contradiction: the recent activity came from a test path, not the production path."

    observation = f"Initial evidence: {evidence}"
    interpretation = (
        "Working hypothesis: the available evidence is sufficient to justify a targeted inspection, "
        "but not sufficient to conclude that the feature is behaving as expected."
    )
    action = "Inspect Feature X and compare the observed path against the expected production path."
    artifact = "inspection_request.json"
    verification = "Inspection requested; result intentionally left open so the environment can provide new information."
    new_information = environment

    event = make_event(
        1, "initial_state", task, observation, interpretation, action, artifact,
        verification, new_information,
        next_iteration_input="Use the inspection result as a constraint on the next hypothesis."
    )
    state = {
        "task": task,
        "evidence": evidence,
        "environment": environment,
        "events": [event],
    }
    return state

def inspect(state):
    if not state or not state.get("events"):
        return {"status": "NO EVIDENCE", "checks": [], "explanation": "Run iteration 1 first."}

    events = state["events"]
    checks = []

    if len(events) < 2:
        checks.append(("Source event exists", True))
        checks.append(("Persistence represented", True))
        checks.append(("Subsequent influence observed", False))
        checks.append(("Correction represented", False))
        return {
            "status": "INCOMPLETE",
            "checks": checks,
            "explanation": "One cycle is evidence of an event, not yet evidence of recursion. Run the next iteration."
        }

    e1, e2 = events[-2], events[-1]
    r1, r2 = e1["recursive"], e2["recursive"]

    checks.append(("Iteration 1 is preserved", bool(e1.get("eventId"))))
    checks.append(("Iteration 2 explicitly references Iteration 1", e2.get("priorEventId") == e1.get("eventId")))
    checks.append(("Iteration 2 uses new information", r2["priorState"] == e1["recursive"]["newInformation"]))
    checks.append(("Action changes after feedback", r2["action"] != r1["action"]))
    checks.append(("Interpretation changes after feedback", r2["interpretation"] != r1["interpretation"]))
    checks.append(("Correction is represented", bool(r2["correction"])))
    checks.append(("Provenance is traceable", bool(e1.get("eventId") and e2.get("eventId"))))

    passed = sum(ok for _, ok in checks)
    status = "INSPECTION PASSED" if passed == len(checks) else "INSPECTION PARTIAL"
    return {
        "status": status,
        "checks": checks,
        "explanation": (
            f"{passed}/{len(checks)} inspection checks passed. "
            "This demonstrates a traceable state transition, not proof of consciousness, autonomy, "
            "or a novel intelligence mechanism."
        )
    }

def render_state(state):
    if not state:
        return "No cycle yet."
    return json.dumps(state, indent=2)

def run_first(task, evidence, environment):
    state = iteration_one(task, evidence, environment)
    return state, render_state(state), inspect(state)

def run_next(state):
    if not state or not state.get("events"):
        return state, render_state(state), inspect(state)

    e1 = state["events"][-1]
    r1 = e1["recursive"]
    new_info = r1["newInformation"]

    # Deterministic correction: the contradiction changes both interpretation and action.
    interpretation = (
        "Correction: the first hypothesis was too broad. The new observation indicates that "
        "the apparent signal may be generated by a test path, so the production path must be "
        "verified before treating the signal as evidence of production behavior."
    )
    action = "Trace the production path, reproduce the signal there, and compare it with the test-path result."
    artifact = "production_path_comparison.json"
    verification = "Second inspection is scoped to the production path and explicitly tests the contradiction."
    subsequent = (
        "Iteration 2 narrows the investigation because Iteration 1's environmental observation "
        "changed the next action."
    )

    e2 = make_event(
        2,
        new_info,
        "Prior cycle produced contradictory environmental evidence.",
        f"Carried forward from Iteration 1: {new_info}",
        interpretation,
        action,
        artifact,
        verification,
        "The next observable should distinguish test-path behavior from production-path behavior.",
        subsequent_influence=subsequent,
        correction="The initial interpretation was narrowed in response to contradictory evidence.",
        next_iteration_input="If reproduction succeeds in production, reassess the original hypothesis with the new trace.",
        prior_event_id=e1["eventId"],
    )

    state = dict(state)
    state["events"] = state["events"] + [e2]
    return state, render_state(state), inspect(state)

def reset():
    return None, "", {"status": "READY", "checks": [], "explanation": "Start with iteration 1."}

def format_inspection(result):
    if not result:
        return "READY"
    lines = [f"### {result['status']}", "", result["explanation"], ""]
    for label, ok in result["checks"]:
        lines.append(f"- {'✅' if ok else '⬜'} {label}")
    return "\n".join(lines)

with gr.Blocks(title=APP_TITLE) as demo:
    gr.Markdown(
        """# 🌀 Recursive Engine Observatory

A tiny, deterministic instrument for inspecting whether **one cycle actually changes the conditions of the next**.

This is deliberately not an autonomous agent. There is no hidden model, no training loop, and no claim of consciousness. The point is to make the recursion **visible, inspectable, and falsifiable**.
"""
    )

    with gr.Row():
        with gr.Column(scale=1):
            task = gr.Textbox(
                label="Task",
                value="Determine whether Feature X should be inspected next.",
                lines=2,
            )
            evidence = gr.Textbox(
                label="Initial evidence",
                value="Feature X has recent activity, but the available evidence is incomplete.",
                lines=3,
            )
            environment = gr.Textbox(
                label="Environmental response / contradiction",
                value="The inspection reveals a contradiction: the recent activity came from a test path, not the production path.",
                lines=4,
            )
            with gr.Row():
                first = gr.Button("▶ Run Iteration 1", variant="primary")
                nxt = gr.Button("↻ Run Next Iteration")
                clear = gr.Button("Reset")

        with gr.Column(scale=1):
            inspection = gr.Markdown(
                "### READY\nRun Iteration 1 to create the first inspectable event.",
                label="Independent inspection",
            )

    gr.Markdown("## Event stream")
    event_json = gr.Code(
        label="Persisted recursive state (session-local in this prototype)",
        language="json",
        lines=24,
    )

    state = gr.State(None)

    # Use a wrapper because the inspection output is structured while Markdown needs text.
    def first_display(task, evidence, environment):
        s = iteration_one(task, evidence, environment)
        return s, render_state(s), format_inspection(inspect(s))

    def next_display(s):
        s2, rendered, result = run_next(s)
        return s2, rendered, format_inspection(result)

    first.click(
        first_display,
        inputs=[task, evidence, environment],
        outputs=[state, event_json, inspection],
        queue=True,
    )
    nxt.click(
        next_display,
        inputs=state,
        outputs=[state, event_json, inspection],
        queue=True,
    )
    clear.click(
        reset,
        outputs=[state, event_json, inspection],
        queue=False,
    )

    gr.Markdown(
        """### What the inspector is looking for

**Source → event → interpretation → implementation/action → observation → correction → subsequent use**

A pattern is interesting only when the chain is traceable. The app intentionally exposes the event IDs and carried-forward state so another person can inspect the transition without relying on the system's own story about itself.
"""
    )

if __name__ == "__main__":
    demo.launch()