# Project export: Uncomfortable

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: UC Berkeley AI Hackathon 2026
- Tagline: For the anxious generation, to become unafraid experimenters and enjoy the process of trial and error.
- Devpost: https://devpost.com/software/uncomfortable
- GitHub: https://github.com/yashishandilya/cal-hacks-26
- Video: https://www.youtube.com/embed/bJqstCiknAQ?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — Yashi Shandilya (1 commits)

## Devpost submission (written by the team)

### Inspiration

Most people quit something or never start, depending on where friction lives in their lives. If a product makes their skin worse for 3 days, they panic, they switch products, or buy into the hype from social media paranoia. Uncomfortable doesn't just block quick pivots; it explains biologically why the discomfort is expected and when it will resolve. The user sees a peer-reviewed research paper linked in-app showing "your skin typically purges on days 5-7 of retinol, you're on day 4, here's what that looks like in the research." This directly solves a real human problem (impatience + fear of failure) with an insight-first design.

### What it does

The anxious generation explores less, aims to avoid any and every feeling of discomfort, and sets up an impossible standard of never slipping. So, when discomfort hits, they interpret it as failure. I built an agent-driven experiment engine that blocks unsafe choices, explains why discomfort is expected, and helps contextualize their concerns with data. It is aimed at helping people experiment and enjoy making sense of the world around them, while taking advantage of an engine rooted in the scientific method of hypothesis building, variable (independent, dependent, control) scoping, and timed trials.

### How we built it

When the user starts an experiment, they get to set their hypothesis, variables (independent, dependent), and the time duration of tracking this. Compiling this into a protocol happens dynamically using Claude by creating high-signal JSON objects from a raw transcript. This forms the Protocol Compiler. Deterministic Validation Engine maps incompatibilities and thresholds using a list for structured output. Reasoning is done by a Research Committee, where we have Research, a De-escalator, and an Arbiter agent. Compaction Agent keeps milestones and recent logs raw, summarizes older prose, and preserves all structured metrics exactly. An A/B harness runs the arbiter on both full and compressed context and asserts the verdict is unchanged. On completion, non-milestone logs are purged (milestones survive); on delete, all Redis keys are removed atomically. All the agents are ten chained and orchestrated by a Master agent. Challenges I ran into Structuring LLM outputs is hard. I used the Instructor library to type-check LLM outputs and enforce JSON schema. It took me some time to realize I was using a deprecated method, which kept returning empty responses despite being syntactically correct. Lots of un-triggered UI calls. Issues with the separation of concerns for agents and constantly chaining agents in a manner that would reset the running context window.

### What we learned

Everything outside of Python and Flask was new for me.

### What's next

I want to add stricter enforcement for experimentation by reducing the time to get an LLM response. Also thinking of a way to visualize milestones and not just all daily logs.

## README (from the GitHub repository)

# cal-hacks-26

## Detected evidence (automated analysis)

Indexed codebase: 13 recognized source files, 110 KB.
- HTML (language) — detected in the code
- Python (language) — detected in the code
- CSS (language) — claimed on Devpost, not found in the code
- Flask (technology) — claimed on Devpost, not found in the code
- Google Gemini (technology) — claimed on Devpost, not found in the code
- JavaScript (language) — claimed on Devpost, not found in the code
- Redis (technology) — claimed on Devpost, not found in the code

## Codebase structure (from repository index)

### Files (21 of 21)

```
.gitignore
app.py
compaction.py
council.py
gc_agent.py
main.py
models.py
protocol_gen.py
protocols/exp_del1782028781-rules.json
protocols/exp_mqnhlri8-rules.json
protocols/exp_mqnhykmr-rules.json
protocols/exp_pantry_777-rules.json
protocols/exp_retinol_mock-rules.json
protocols/exp_skin_999-rules.json
protocols/exp_test1782028385-rules.json
README.md
runtime.py
store.py
templates/index.html
tracing.py
validation.py
```

### Dependencies

No dependency index available.

### Recent commits (newest first)

- cleaned up directory-folder
- cleaned up directory
- added better ui flows
- cleaned up commits
- adding council, arize tracing and mock data using seed.py
- some test scripts
- garbage collector + compaction
- setting up redis agent memory
- validation for gt, lt, etc operators befoe we move forward with reasoning with logs
- Merge branch 'main' of github.com:yashishandilya/cal-hacks-26
- sets up data model using pydantic, creates a schema for deterministic validator, generates protocol per experiment
- Initial commit

## Key source files (fetched from GitHub, selected and truncated for size)

### main.py

```python
from datetime import datetime, timezone
from enum import Enum
from typing import TypeVar, Generic, List, Dict, Any, Optional
from pydantic import BaseModel, Field

class experimentState(BaseModel):
    active: bool = False
    queued: bool = False
    restore: bool = False
    kill: bool = False

class userVar(BaseModel):
    varId: str
    varName: str
    constraints: Dict[str, Any] = Field(default_factory=dict)

class variableTriad(BaseModel):
    indVar: userVar
    depVar: Dict[str, Any]
    conVar: list[userVar]

class dailyLogEntry(BaseModel):
    
    expId: str
    dateTime: datetime = Field(default_factory=lambda : datetime.now(timezone.utc))
    milestone: bool = False
    expStatus: experimentState
    payload: Dict[str, Any]
    chatTranscript: Optional[str] = None

class Experiment(BaseModel):
    expId: str
    dateTime: datetime = Field(default_factory=lambda : datetime.now(timezone.utc))
    protocol: str
    varTriad: variableTriad
    compromised: bool = False
    logs: List[dailyLogEntry] = Field(default_factory=list)

if __name__ == "__main__":
    print("="*60)
    print("VALIDATION TESTS")
    print("="*60 + "\n")

    # This raw dictionary simulates what your Setup Agent outputs after a chat session
    llm_simulated_setup = {
        "expId": "exp_frequency_vector_01",
        "protocol": "Isolating Retinol Frequency Bounds (PM ONLY)",
        "expStatus": {"active": True, "queued": False},
        "varTriad": {
            "indVar": {"varId": "v_retinol_05", "varName": "Retinol 0.5% Serum"},
            "conVar": [
                {"varId": "c_wash", "varName": "Gentle Wash"},
                {"varId": "c_cream", "varName": "Barrier Cream"}
            ],
            # Enforces dynamic boundaries of whatever parameter criteria user wants to isolate
            "depVar": {
                "type": "object",
                "properties": {
                    "redness": {"type": "integer", "minimum": 1, "maximum": 10},
                    "tightness": {"type": "integer", "minimum": 1, "maximum": 10}
                },
                "required": ["redness", "tightness"]
            }
        }
    }

    # Initialize the engine artifact state object
    active_study = Experiment(**llm_simulated_setup)
    print("Model Compilation Successful: Standalone baseline configuration generated.\n")

    # Ingest a clean, valid day log 
    sample_day_log = dailyLogEntry(
        expId=active_study.expId,
        expStatus=experimentState(active=True, queued=False),
        payload={"redness": 3, "tightness": 2},
        chatTranscript="Applied at 10 PM. Smooth consistency, zero dynamic flare-ups."
    )
    active_study.logs.append(sample_day_log)
    print(f"Day 1 Transaction verified and nested at timestamp: {sample_day_log.dateTime}")

    # Final persistent storage payload serialization lookup
    print("\n" + "="*60)
    print("STORAGE DOCUMENT DUMP")
    print("="*60)
    print(active_study.model_dump_json(indent=2))
```

### app.py

```python
"""
Flask web layer (Chunk E). Serves the Experimenter mockup and exposes JSON endpoints
that wire the UI to the Master orchestrator. The camera/environmental/sub-experiment
panels stay as static mock UI; only the journal -> runTick slice is real.
"""

import json

from flask import Flask, render_template, jsonify, request

import store
from runtime import runTick
from council import researchTopic
from compaction import compactExperiment
from protocol_gen import generate_dynamic_protocol, normalizeMetricKey
from models import ProtocolSchema
from main import Experiment, variableTriad, userVar

app = Flask(__name__)


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/api/experiments")
def listExperiments():
    out = []
    for expId in store.listExperimentIds():
        exp = store.getExperiment(expId)
        out.append({
            "expId": expId,
            "protocol": exp.protocol if exp else "",
            "logCount": len(exp.logs) if exp else 0,
        })
    return jsonify(out)


@app.route("/api/experiments/<expId>/logs")
def getLogs(expId):
    logs = store.getLogs(expId)
    return jsonify([log.model_dump(mode="json") for log in logs])


# Runs one daily log through the Master orchestrator and returns the full PipelineTrace:
# verdict, violations, council verdict, de-escalation + recovery, and the stage trace.
@app.route("/api/experiments/<expId>/log", methods=["POST"])
def postLog(expId):
    transcript = (request.get_json(force=True) or {}).get("transcript", "").strip()
    if not transcript:
        return jsonify({"error": "transcript is required"}), 400
    trace = runTick(expId, transcript)
    return jsonify(trace.model_dump(mode="json"))


# Runs the Researcher on a query (defaults to the experiment's protocol topic) and returns
# real grounded findings + cited web sources for the right-hand Research sources panel.
@app.route("/api/experiments/<expId>/research", methods=["POST"])
def research(expId):
    query = (request.get_json(force=True) or {}).get("query", "").strip()
    if not query:
        exp = store.getExperiment(expId)
        query = exp.protocol if exp else expId
    return jsonify(researchTopic(query))


# Returns the compaction token stats (Token Company readout). Drops the bulky compressed
# context so the response stays small for the UI badge.
@app.route("/api/experiments/<expId>/compaction")
def compaction(expId):
    r = compactExperiment(expId)
    return jsonify({k: r[k] for k in ("tokensBefore", "tokensAfter", "reductionRatio", "logsCompacted", "logsKeptRaw")})


@app.route("/api/experiments/<expId>/protocol")
def getProtocol(expId):
    raw = store.getProtocol(expId)
    if raw is None:
        return jsonify({"error": "no protocol cached"}), 404
    return app.response_class(raw, mimetype="application/json")


# Compiles the Setup page's config cards into a real ProtocolSchema. The frontend
# assembles the hypothesis, tracked metrics, window, and committee into one transcript;
# this runs the deterministic protocol compiler (Gemini-backed), which caches the
# compiled rulebook in Redis and writes protocols/<expId>-rules.json. Returns the
# freshly cached protocol JSON so the right-hand spec panel can redraw immediately.
@app.route("/api/experiments/<expId>/compile", methods=["POST"])
def compileProtocol(expId):
    body = request.get_json(force=True) or {}
    transcript = (body.get("transcript") or "").strip()
    hypothesis = (body.get("hypothesis") or "").strip()
    # The Watch-for editor: what the user monitors (the input they change) maps to the
    # independent variable; the result they want (the outcome they measure) maps to the
    # dependent variable. 'metrics' is accepted as a legacy alias for dependent.
    independent = body.get("independent") or []
    dependent = body.get("dependent") or body.get("metrics") or []
    if not transcript:
        return jsonify({"error": "transcript is required"}), 400

    try:
        generate_dynamic_protocol(transcript, expId)
    except Exception as e:
        return jsonify({"error": str(e)}), 500

    raw = store.getProtocol(expId)
    if raw is None:
        return jsonify({"error": "compiled protocol was not cached"}), 500
    protocol = ProtocolSchema.model_validate_json(raw)

    # Rebuild the variable triad so the right-hand spec reflects the Watch-for editor.
    # The user's "what I monitor" is the independent variable, "what result I want" is the
    # dependent variable; anything they left blank falls back to the variables the protocol
    # compiler extracted.
    triad = buildVariableTriad(protocol, independent, dependent, hypothesis)

    # Persist the new triad onto the Experiment record (creating it if this is the first
    # compile). Logs live under their own Redis key, so we keep the header's logs empty.
    exp = store.getExperiment(expId)
    if exp is None:
        exp = Experiment(expId=expId, protocol=protocol.protocol, varTriad=triad)
    else:
        exp.protocol = protocol.protocol
        exp.varTriad = triad
    exp.logs = []
    store.saveExperiment(exp)
    # Note: do NOT even think of clearing the logs list here. A brand-new experiment has no logs yet (so it
    # starts empty on its own), and recompiling an existing experiment should keep its journal.

    # Live research grounded on the hypothesis, so the Research sources panel reacts too.
    # Guarded: a research failure shouldn't void an otherwise-successful compile.
    research = {"findings": "", "sources": []}
    try:
        research = researchTopic(hypothesis or protocol.protocol)
    except Exception as e:
        research = {"findings": "", "sources": [], "error": str(e)}

    return jsonify({"protocol": json.loads(raw), "research": research})


# Builds a variableTriad from the user's Watch-for editor, with the compiled protocol as
# a fallback. The user's monitored inputs become the independent variable (first one) +
# controls (the rest); the results they wa
[truncated — 3547 more characters]
```

### gc_agent.py

```python
"""
Garbage Collector (Token Company track, storage side).

Two lifecycle actions over Redis:
- purge-non-milestone-on-complete: when an experiment ends, keep only milestone logs.
- delete-on-delete: remove an experiment and all its data (reuses store.deleteExperiment).
Compaction (the rolling-summary side) lives in compaction.py.
"""

import store


# Drops every non-milestone log for an experiment, keeping only the milestones, and
# reports how many were removed. Used when an experiment completes and the day-to-day
# noise is no longer needed but the key events must survive.
def purgeNonMilestones(expId: str) -> dict:
    logs = store.getLogs(expId)
    kept = [log for log in logs if log.milestone]
    store.replaceLogs(expId, kept)
    return {"before": len(logs), "kept": len(kept), "removed": len(logs) - len(kept)}


# Marks an experiment finished by purging its non-milestone logs. Kept as its own name so
# callers express intent ("this experiment is over") rather than the mechanism.
def completeExperiment(expId: str) -> dict:
    return purgeNonMilestones(expId)


# Hard-deletes an experiment and all its data when the user deletes it. Thin wrapper over
# the store so the GC is the single place lifecycle cleanup is expressed.
def deleteExperimentData(expId: str) -> int:
    return store.deleteExperiment(expId)

```

### validation.py

```python
import json
from typing import Any, List, Dict, Union
from models import ProtocolSchema, ComparisonOperator

class ValidationEngine:
    def __init__(self, rules: "Union[str, ProtocolSchema]"):
        # Accept either a path to a rules JSON file (read from disk) or an already-loaded
        # ProtocolSchema (used by the Master orchestrator, which holds the Redis-cached protocol).
        if isinstance(rules, ProtocolSchema):
            self.rules = rules
        else:
            with open(rules, 'r') as file:
                # Re-hydrates the saved file directly using your ProtocolSchema structure
                self.rules = ProtocolSchema.model_validate_json(file.read())

        # Functional dispatcher mapping every single one of your custom operators
        self.operators = {
            ComparisonOperator.GT: lambda dataVal, ruleLimit: float(dataVal) > float(ruleLimit),
            ComparisonOperator.GTE: lambda dataVal, ruleLimit: float(dataVal) >= float(ruleLimit),
            ComparisonOperator.LT: lambda dataVal, ruleLimit: float(dataVal) < float(ruleLimit),
            ComparisonOperator.LTE: lambda dataVal, ruleLimit: float(dataVal) <= float(ruleLimit),
            ComparisonOperator.EQ: lambda dataVal, ruleLimit: str(dataVal) == str(ruleLimit),
            ComparisonOperator.NOT_EQ: lambda dataVal, ruleLimit: str(dataVal) != str(ruleLimit),
            ComparisonOperator.CONTAINS: lambda dataVal, ruleLimit: str(ruleLimit) in str(dataVal),
            ComparisonOperator.NO_CONTAINS: lambda dataVal, ruleLimit: str(ruleLimit) not in str(dataVal),
        }

    def validateAction(self, newActionId: str, activeStack: List[str]) -> tuple:
        """Evaluates O(1) compound conflicts using your incompatibilities matrix mapping."""
        conflicts = self.rules.incompatibilities.get(newActionId, [])
        for item in conflicts:
            if item in activeStack:
                return False, f"Conflict detected: {newActionId} cannot be used with {item}"
        return True, "Valid"

    def checkThreshold(self, metricKey: str, value: Any) -> tuple:
        """Scans your thresholds list sequentially to verify structural state constraints."""
        for threshold in self.rules.thresholds:
            if threshold.metricKey == metricKey:
                operator = ComparisonOperator(threshold.operator)
                isBreached = self.operators[operator](value, threshold.limit)
                if isBreached:
                    return False, threshold.errorMessage
        return True, "Within limits"
    
    
```

### tracing.py

```python
"""
Arize Phoenix tracing (observability track).

Launches a local Phoenix server and auto-instruments our LLM calls (Gemini via the
google-genai SDK and instructor), so every agent step shows up as a trace at
http://localhost:6006. Call startTracing() once at app startup.
"""

import sys
import phoenix as px
from phoenix.otel import register

# Phoenix prints an emoji banner on launch; the Windows console (cp1252) can't encode it
# and crashes. Force stdout/stderr to UTF-8 so the launch never dies on the banner.
try:
    sys.stdout.reconfigure(encoding="utf-8")
    sys.stderr.reconfigure(encoding="utf-8")
except Exception:
    pass

_started = False


# Launches the local Phoenix UI and wires OpenTelemetry auto-instrumentation so all
# Gemini/instructor calls are captured as spans. Idempotent: safe to call more than once.
def startTracing():
    global _started
    if _started:
        return
    px.launch_app()
    # auto_instrument is off: the openinference google-genai instrumentor is incompatible
    # with our google-genai version. We emit our own spans (see runtime.py) instead, which
    # also gives a cleaner tick -> ingest/safety/council trace tree.
    register(project_name="the-experimenter", auto_instrument=False)
    _started = True
    print("[tracing] Phoenix live at http://localhost:6006")


# Fetches recent traces for one experiment from Phoenix: each runTick root span carries
# the expId, and its child spans (ingest/safety/council) share the trace_id. Returns the
# most recent traces with their spans + per-span latency for the in-app trace panel.
def getExperimentTraces(expId: str, limit: int = 10) -> list:
    import pandas as pd
    from phoenix.client import Client

    df = Client(base_url="http://localhost:6006").spans.get_spans_dataframe(
        project_identifier="the-experimenter"
    )
    if df is None or len(df) == 0 or "attributes.expId" not in df.columns:
        return []

    roots = df[df["attributes.expId"] == expId].sort_values("start_time", ascending=False).head(limit)

    def cell(row, col):
        val = row.get(col)
        return None if val is None or pd.isna(val) else val

    out = []
    for _, root in roots.iterrows():
        traceId = root["context.trace_id"]
        spansDf = df[df["context.trace_id"] == traceId].sort_values("start_time")
        spans = []
        for _, s in spansDf.iterrows():
            latency = None
            if not pd.isna(s["start_time"]) and not pd.isna(s["end_time"]):
                latency = round((s["end_time"] - s["start_time"]).total_seconds() * 1000, 1)
            spans.append({
                "name": s["name"],
                "latencyMs": latency,
                "verdict": cell(s, "attributes.verdict"),
                "councilVerdict": cell(s, "attributes.councilVerdict"),
                "detail": cell(s, "attributes.detail"),
            })
        out.append({
            "traceId": traceId,
            "startTime": str(root["start_time"]),
            "verdict": cell(root, "attributes.verdict"),
            "spans": spans,
        })
    return out

```

### models.py

```python
from enum import Enum
from datetime import datetime, timezone
from typing import TypeVar, Generic, List, Dict, Any, Optional, Union
from pydantic import BaseModel, Field

class ComparisonOperator(str, Enum):
    GT = "gt"
    GTE = "gte"
    LT = "lt"
    LTE = "lte"
    EQ = "eq"
    NOT_EQ = "neq"
    CONTAINS = "contains" 
    NO_CONTAINS = "does not contains"


class Threshold(BaseModel):
    metricKey: str
    operator: str
    limit: "Union[float, str]" # Allows for numeric or categorical comparisons
    errorMessage: str

class ProtocolSchema(BaseModel):
    protocol: str
    homeostasis_lockout_days: int
    incompatibilities: Dict[str, List[str]] # e.g., {"v_a": ["v_b", "v_c"]}
    thresholds: List[Threshold]


# Normalized output of the Orchestrator's ingest step. The normalization call is
# handed the protocol's own metricKeys / variable IDs, so it can ONLY emit keys
# that already exist in the protocol. This closes the seam where free-text chat
# logs produced keys that never matched the protocol and silently passed safety.
class TelemetryPacket(BaseModel):
    metrics: Dict[str, Any]          # metricKey -> measured value, keyed to the protocol
    actions: List[str]               # variable IDs applied/used in this log entry
    notes: Optional[str] = None      # anything the normalizer couldn't map to a key


# Raised by the Orchestrator when the deterministic ValidationEngine flags a clash
# or a breached threshold. Carries the list of human-readable violation messages so
# the trace envelope and the UI banner can surface exactly what tripped, and so the
# Orchestrator can block the Redis write when this is raised.
class ProtocolViolationException(Exception):
    def __init__(self, violations: List[str]):
        self.violations = violations
        super().__init__("; ".join(violations))


# One step in the Master orchestrator's run, recorded for the UI trace ledger:
# which sub-step ran, how long it took, and a short human-readable result line.
class TraceStage(BaseModel):
    agent: str               # name of the step, e.g. "ingest" or "safety"
    durationMs: float        # how long this step took, in milliseconds
    summary: str             # short readable description of what this step produced


# The full envelope the Master orchestrator returns for one log. The frontend reads
# 'verdict' + 'violations' for the warning banner and 'stages' for the trace ledger.
class PipelineTrace(BaseModel):
    expId: str
    timestamp: str                              # ISO timestamp of the run
    transcript: str                             # the raw log text that was processed
    verdict: str                                # "ok" if accepted, "blocked" if a violation stopped it
    stages: List[TraceStage] = Field(default_factory=list)
    violations: List[str] = Field(default_factory=list)
    logStored: bool = False                     # whether the log was actually written to Redis
    # Council outputs (added after the deterministic gate). councilVerdict is the arbiter's
    # continue/adjust/stop call (distinct from verdict above, which is ok/blocked); the
    # de-escalation fields are populated only when blocked.
    councilVerdict: Optional[str] = None
    deEscalationMessage: Optional[str] = None
    recoverySteps: List[str] = Field(default_factory=list)

    
```

### council.py

```python
"""
Committee / Council agents (running on Gemini for now).

D1: Researcher - uses Gemini's Google Search grounding to pull real, cited sources
that ground the committee's reasoning. Feeds the UI "research sources" panel.
"""

import os
import instructor
from typing import List
from pydantic import BaseModel
from google import genai
from google.genai import types
from dotenv import load_dotenv

load_dotenv()
geminiClient = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))


# Domains we count as academic/peer-reviewed: literature indexes, journal publishers, and
# university (.edu / .ac.) sites. The grounding chunk's title is the source domain (e.g.
# "ncbi.nlm.nih.gov"), so a substring check against this list filters out blogs and
# product/marketing pages.
ACADEMIC_DOMAINS = (
    "ncbi.nlm.nih.gov", "pubmed", "pmc", "doi.org", ".edu", ".ac.", "nature.com",
    "sciencedirect", "springer", "wiley", "onlinelibrary", "tandfonline", "sagepub",
    "jamanetwork", "nejm.org", "thelancet", "bmj.com", "cell.com", "frontiersin",
    "mdpi.com", "plos.org", "cochrane", "researchgate", "semanticscholar", "scholar.google",
    "oup.com", "cambridge.org", "annualreviews", "karger", "dovepress", "jaad.org", "elsevier",
)


def isAcademicSource(title: str, url: str) -> bool:
    haystack = f"{title or ''} {url or ''}".lower()
    return any(domain in haystack for domain in ACADEMIC_DOMAINS)


# Researches an experiment topic with Gemini's Google Search grounding, returning a short
# findings summary plus only the peer-reviewed/academic sources Gemini cited (title + url).
def researchTopic(query: str) -> dict:
    prompt = (
        "Research this experiment topic using ONLY peer-reviewed academic and scientific "
        "sources: journal articles, clinical trials, systematic reviews, PubMed/PMC entries, "
        "and university (.edu) publications. Do NOT use blogs, news articles, product pages, "
        "or commercial/marketing sites. Give a brief, factual 2-3 sentence summary of what "
        f"the peer-reviewed evidence says. Topic: {query}"
    )
    config = types.GenerateContentConfig(
        tools=[types.Tool(google_search=types.GoogleSearch())]
    )
    response = geminiClient.models.generate_content(
        model="gemini-2.5-flash", contents=prompt, config=config
    )

    findings = (response.text or "").strip()

    # Pull cited sources from the grounding metadata, then keep only academic ones. Guards
    # for when Gemini returns no grounding at all (then we keep findings + an empty list).
    sources: List[dict] = []
    candidate = response.candidates[0] if response.candidates else None
    metadata = getattr(candidate, "grounding_metadata", None) if candidate else None
    chunks = getattr(metadata, "grounding_chunks", None) if metadata else None
    if chunks:
        for chunk in chunks:
            web = getattr(chunk, "web", None)
            if web and isAcademicSource(web.title, web.uri):
                sources.append({"title": web.title, "url": web.uri})

    return {"query": query, "findings": findings, "sources": sources}


# The De-escalator's output: a calm, plain-language explanation of what tripped and a
# few concrete recovery steps. Maps to the UI's red "De-escalator triggered" flag box.
class DeEscalation(BaseModel):
    message: str               # calm explanation of what happened and why it matters
    recoverySteps: List[str]   # 2-4 concrete steps to recover safely


# Given the safety violations (and optional context), produces a supportive de-escalation
# instead of a raw error. Runs only after the deterministic gate has already blocked.
def deEscalate(violations: List[str], context: str = "") -> DeEscalation:
    client = instructor.from_provider("google/gemini-2.5-flash", api_key=os.getenv("GEMINI_API_KEY"))
    joined = "; ".join(violations)
    return client.create(
        model="gemini-2.5-flash",
        response_model=DeEscalation,
        messages=[
            {"role": "system", "content": (
                "You are a calm, reassuring safety de-escalator for a personal-experiment app. A "
                "protocol violation was detected and the action was blocked. Explain plainly and "
                "without alarm what happened and why it matters, then give 2-4 concrete recovery "
                "steps. Be supportive, never preachy."
            )},
            {"role": "user", "content": f"Violations: {joined}\nContext: {context}"},
        ],
        max_retries=3,
    )

```

### store.py

```python
"""
Redis-backed state store. Sole datastore for the experiment engine.

Key layout
----------
exp:{expId}              -> Experiment JSON (string)
exp:{expId}:logs         -> Redis LIST of dailyLogEntry JSON (oldest -> newest)
exp:{expId}:protocol     -> compiled ProtocolSchema JSON (cache for sub-ms reads)
experiments              -> SET of all known expIds (index)

Everything the Orchestrator, Garbage Collector/Compaction agent, and Council
read or write goes through this module. No other file talks to Redis directly.
"""

import os
import redis
from typing import List, Optional
from dotenv import load_dotenv

# Reuse the experiment models already defined in main.py rather than redefining.
from main import Experiment, dailyLogEntry

load_dotenv()

redisUrl = os.getenv("REDIS_URL")
if not redisUrl:
    raise RuntimeError(
        "REDIS_URL is missing. Add your Redis Essentials connection string to "
        ".env, e.g. REDIS_URL=redis://default:<password>@<host>:<port>"
    )

# decode_responses=True so every read comes back as str instead of raw bytes.
redisClient = redis.from_url(redisUrl, decode_responses=True)


# Cheap connectivity check. Called once at orchestrator startup so a bad
# REDIS_URL fails loudly up front instead of mid-pipeline.
def ping() -> bool:
    return redisClient.ping()


# Builds the key that holds the Experiment header document for an experiment.
def expKey(expId: str) -> str:
    return f"exp:{expId}"


# Builds the key that holds the Redis LIST of daily log entries.
def logsKey(expId: str) -> str:
    return f"exp:{expId}:logs"


# Builds the key that holds the cached compiled protocol JSON string.
def protocolKey(expId: str) -> str:
    return f"exp:{expId}:protocol"


# Upserts the Experiment header (everything except the logs list, which lives
# under its own key). Also registers the expId in the master 'experiments' set
# so the Garbage Collector and UI can enumerate every experiment.
def saveExperiment(exp: Experiment) -> None:
    pipe = redisClient.pipeline()
    pipe.set(expKey(exp.expId), exp.model_dump_json())
    pipe.sadd("experiments", exp.expId)
    pipe.execute()


# Fetches an Experiment by id and re-hydrates its logs from the separate list
# key onto the object. Returns None if the experiment does not exist.
def getExperiment(expId: str) -> Optional[Experiment]:
    raw = redisClient.get(expKey(expId))
    if raw is None:
        return None
    exp = Experiment.model_validate_json(raw)
    exp.logs = getLogs(expId)
    return exp


# Returns every known experiment id, sorted, from the master index set.
def listExperimentIds() -> List[str]:
    return sorted(redisClient.smembers("experiments"))


# Hard-deletes an experiment: header, logs list, cached protocol, and its entry
# in the master set. Used by the Garbage Collector on experiment delete.
# Returns how many of the three data keys actually existed and were removed.
def deleteExperiment(expId: str) -> int:
    pipe = redisClient.pipeline()
    pipe.delete(expKey(expId))
    pipe.delete(logsKey(expId))
    pipe.delete(protocolKey(expId))
    pipe.srem("experiments", expId)
    results = pipe.execute()
    return sum(results[:3])


# Appends one daily log entry to the experiment's log list. Returns the new
# total log count for that experiment.
def appendLog(entry: dailyLogEntry) -> int:
    return redisClient.rpush(logsKey(entry.expId), entry.model_dump_json())


# Returns all daily log entries for an experiment, oldest first.
def getLogs(expId: str) -> List[dailyLogEntry]:
    raw = redisClient.lrange(logsKey(expId), 0, -1)
    return [dailyLogEntry.model_validate_json(item) for item in raw]


# Atomically overwrites the entire logs list for an experiment. Used by the
# Compaction agent after folding old non-milestone logs into a summary entry,
# and by the purge-non-milestone-on-complete path.
def replaceLogs(expId: str, entries: List[dailyLogEntry]) -> None:
    pipe = redisClient.pipeline()
    pipe.delete(logsKey(expId))
    if entries:
        pipe.rpush(logsKey(expId), *[e.model_dump_json() for e in entries])
    pipe.execute()


# Caches the compiled protocol string so the Council/arbiter can read it in
# sub-ms time instead of re-reading the protocols/*.json file from disk.
def cacheProtocol(expId: str, protocolJson: str) -> None:
    redisClient.set(protocolKey(expId), protocolJson)


# Returns the cached compiled protocol JSON string, or None if not cached yet.
def getProtocol(expId: str) -> Optional[str]:
    return redisClient.get(protocolKey(expId))

```

### protocol_gen.py

```python
import os
import re
import time
import instructor
from typing import List
from pydantic import BaseModel
from google import genai
from models import ProtocolSchema, Threshold
from dotenv import load_dotenv

# Reuse the Redis store so a freshly compiled protocol can be cached for inspection.
import store


# One conflict rule expressed as a flat key/list pair. We ask the LLM for these as a
# LIST because Gemini's structured output returns {} for an open {str: [str]} dict
# (the same failure mode that left incompatibilities empty before); a list fills correctly.
class IncompatibilityRule(BaseModel):
    variableId: str
    clashesWith: List[str]


# The model-facing draft of a protocol. Identical to ProtocolSchema except the
# incompatibilities are a list of rules instead of an open dict; we convert it to a
# real ProtocolSchema (dict-shaped) in code so the ValidationEngine is unaffected.
class ProtocolDraft(BaseModel):
    protocol: str
    homeostasis_lockout_days: int
    incompatibilities: List[IncompatibilityRule]
    thresholds: List[Threshold]


# Forces a metricKey into a stable snake_case ID: lowercase, every run of non
# alphanumeric characters collapsed to one underscore, and a 'v_' prefix guaranteed
# (e.g. "clinical redness score" -> "v_clinical_redness_score").
def normalizeMetricKey(raw: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "_", raw.strip().lower()).strip("_")
    if not slug.startswith("v_"):
        slug = "v_" + slug
    return slug


def generate_dynamic_protocol(userTranscript: str, experimentId: str) -> str:
    startTime = time.perf_counter()
    load_dotenv()
    apiKey = os.getenv("GEMINI_API_KEY")

    if not apiKey:
        raise RuntimeError("GEMINI_API_KEY is missing. Add it to your .env file before running the protocol generator.")

    print(f"[protocol_gen] Environment loaded in {time.perf_counter() - startTime:.2f}s")
    client = instructor.from_provider("google/gemini-2.5-flash", api_key=apiKey)
    
    systemInstruction = """
    You are a Deterministic Protocol Compiler. Translate user experimental 
    goals into the strict JSON schema provided.
    1. Extract all experimental variables as 'v_...' IDs.
    2. Normalize all conflict logic into the 'incompatibilities' map.
    3. Define safety thresholds strictly using the provided operators.
    4. Never invent keys on your own. Always use Protocol Schema.
    5. All variable names must be normalized IDs (e.g., lowercase with underscores like v_item_name).
    6. 'incompatibilities' is a LIST of rules. Add one rule for EVERY conflict the user mentions:
       each rule has 'variableId' (the item) and 'clashesWith' (the list of variable IDs it must not be combined with).
       If the user says "A cannot be used with B", add a rule {variableId: A, clashesWith: [B]}.
    7. For each item inside the thresholds list, carefully populate 'metricKey', 'operator', 'limit', and 'errorMessage'.
       'metricKey' MUST be a normalized snake_case identifier prefixed with 'v_' (lowercase, words joined by
       underscores, no spaces), e.g. 'v_redness_score'. Never use a free-text phrase with spaces.
    8. The 'operator' field MUST strictly match one of these exact ComparisonOperator values: 'gt', 'gte', 'lt', 'lte', 'eq', 'neq', 'contains', 'does not contains'.
    """

    print("[protocol_gen] Sending request to Gemini...")
    requestStartTime = time.perf_counter()
    draft = client.create(
        model="gemini-2.5-flash",
        response_model=ProtocolDraft,
        messages=[
            {"role": "system", "content": systemInstruction},
            {"role": "user", "content": f"Compile this transcript: {userTranscript}"}
        ],
        max_retries=3
    )
    print(f"[protocol_gen] Gemini response received in {time.perf_counter() - requestStartTime:.2f}s")

    # Deterministic guard: re-normalize every metricKey in code so it is a stable v_ ID
    # no matter how the model phrased it, mirroring the prompt rule above.
    for threshold in draft.thresholds:
        threshold.metricKey = normalizeMetricKey(threshold.metricKey)

    # Fold the list of conflict rules back into the flat dict the ValidationEngine reads.
    incompatibilities = {rule.variableId: rule.clashesWith for rule in draft.incompatibilities}
    protocol = ProtocolSchema(
        protocol=draft.protocol,
        homeostasis_lockout_days=draft.homeostasis_lockout_days,
        incompatibilities=incompatibilities,
        thresholds=draft.thresholds,
    )

    os.makedirs("protocols", exist_ok=True)
    file_path = f"protocols/{experimentId}-rules.json"

    protocolJson = protocol.model_dump_json(indent=2)
    with open(file_path, "w") as f:
        f.write(protocolJson)

    # Cache the compiled protocol in Redis under the experiment id so it can be
    # inspected directly (Redis Insight / CLI) and read sub-ms by later agents.
    store.cacheProtocol(experimentId, protocolJson)

    print(f"[protocol_gen] Wrote protocol JSON + cached to Redis in {time.perf_counter() - startTime:.2f}s total")

    return file_path
```

[2 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]