Skip to content
9 changes: 9 additions & 0 deletions codeframe/core/proof/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,11 @@
from typing import Optional


# Shared filename for the per-workspace PROOF9 config (issue #556).
# Lives under workspace.state_dir.
PROOF_CONFIG_FILENAME = "proof_config.json"


class Gate(str, Enum):
"""The 9 proof gates."""

Expand All @@ -24,6 +29,10 @@ class Gate(str, Enum):
MANUAL = "manual"


# Canonical PROOF9 gate order (matches the Gate enum declaration).
PROOF9_GATE_ORDER: tuple[str, ...] = tuple(g.value for g in Gate)


class GlitchType(str, Enum):
"""Classification of the glitch that produced a requirement."""

Expand Down
57 changes: 55 additions & 2 deletions codeframe/core/proof/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,19 +5,57 @@
and attaches evidence artifacts.
"""

import json
import logging
import uuid
from datetime import datetime, timezone
from typing import Optional

from codeframe.core.proof import ledger
from codeframe.core.proof.evidence import attach_evidence
from codeframe.core.proof.models import Gate, ProofRun, ReqStatus
from codeframe.core.proof.models import (
PROOF_CONFIG_FILENAME,
Gate,
ProofRun,
ReqStatus,
)
from codeframe.core.proof.scope import get_changed_scope, intersects
from codeframe.core.workspace import Workspace

logger = logging.getLogger(__name__)


def _load_proof_config(workspace: Workspace) -> tuple[Optional[set[Gate]], str]:
"""Load (enabled_gates, strictness) from .codeframe/proof_config.json.

Returns:
(enabled_gates, strictness). enabled_gates is None when no config file
exists (meaning "all gates allowed"); a set of Gate enums otherwise.
strictness defaults to 'strict' when missing or invalid.
"""
path = workspace.state_dir / PROOF_CONFIG_FILENAME
if not path.exists():
return None, "strict"
try:
data = json.loads(path.read_text())
except (OSError, json.JSONDecodeError) as exc:
logger.warning("Invalid %s — using defaults: %s", PROOF_CONFIG_FILENAME, exc)
return None, "strict"

enabled: Optional[set[Gate]] = None
gates_raw = data.get("enabled_gates")
if isinstance(gates_raw, list):
Comment on lines +39 to +47
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Handle non-object JSON configs to avoid runtime crashes

At Line 46, data.get(...) assumes the parsed JSON is a dict. If proof_config.json is valid JSON but not an object (e.g., []), this raises AttributeError and aborts the run instead of falling back to defaults.

Suggested fix
     try:
         data = json.loads(path.read_text())
     except (OSError, json.JSONDecodeError) as exc:
         logger.warning("Invalid %s — using defaults: %s", PROOF_CONFIG_FILENAME, exc)
         return None, "strict"
+    if not isinstance(data, dict):
+        logger.warning(
+            "Invalid %s shape (expected JSON object) — using defaults",
+            PROOF_CONFIG_FILENAME,
+        )
+        return None, "strict"
 
     enabled: Optional[set[Gate]] = None
     gates_raw = data.get("enabled_gates")
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
try:
data = json.loads(path.read_text())
except (OSError, json.JSONDecodeError) as exc:
logger.warning("Invalid %s — using defaults: %s", PROOF_CONFIG_FILENAME, exc)
return None, "strict"
enabled: Optional[set[Gate]] = None
gates_raw = data.get("enabled_gates")
if isinstance(gates_raw, list):
try:
data = json.loads(path.read_text())
except (OSError, json.JSONDecodeError) as exc:
logger.warning("Invalid %s — using defaults: %s", PROOF_CONFIG_FILENAME, exc)
return None, "strict"
if not isinstance(data, dict):
logger.warning(
"Invalid %s shape (expected JSON object) — using defaults",
PROOF_CONFIG_FILENAME,
)
return None, "strict"
enabled: Optional[set[Gate]] = None
gates_raw = data.get("enabled_gates")
if isinstance(gates_raw, list):
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@codeframe/core/proof/runner.py` around lines 39 - 47, The code assumes parsed
JSON "data" is a dict and calls data.get(...), which will crash if the JSON root
is not an object; modify the logic in runner.py (around the variables data,
gates_raw, enabled_gates and PROOF_CONFIG_FILENAME) to first verify
isinstance(data, dict) before using data.get, and if not a dict treat the config
as invalid (log a warning similar to the existing one) and fall back to the
defaults (return None, "strict" or the existing default path) so enabled remains
None and downstream code won’t raise AttributeError.

enabled = set()
valid_values = {g.value for g in Gate}
for name in gates_raw:
if name in valid_values:
enabled.add(Gate(name))

strictness = data.get("strictness", "strict")
if strictness not in ("strict", "warn"):
strictness = "strict"
return enabled, strictness

# Map PROOF9 gates to existing core/gates.py gate names
_GATE_TO_CORE: dict[Gate, str] = {
Gate.UNIT: "pytest",
Expand Down Expand Up @@ -71,6 +109,9 @@ def run_proof(

started_at = datetime.now(timezone.utc)

# Load PROOF9 config (enabled gates + strictness)
enabled_gates, strictness = _load_proof_config(workspace)

# Expire any stale waivers
expired = ledger.check_expired_waivers(workspace)
if expired:
Expand Down Expand Up @@ -117,6 +158,10 @@ def run_proof(
if gate_filter and obl.gate != gate_filter:
continue

# Apply config-driven gate filter (None means "all allowed")
if enabled_gates is not None and obl.gate not in enabled_gates:
continue

# Run the gate
passed, output = _run_gate(workspace, obl.gate)

Expand Down Expand Up @@ -146,7 +191,15 @@ def run_proof(
completed_at = datetime.now(timezone.utc)
duration_ms = int((completed_at - started_at).total_seconds() * 1000)
executed = [passed for gate_results in results.values() for _, passed in gate_results]
overall_passed = all(executed) if executed else True
all_passed = all(executed) if executed else True
if not all_passed and strictness == "warn":
logger.warning(
"Proof run %s had gate failures but strictness='warn' — overall_passed=True",
run_id,
)
overall_passed = True
else:
overall_passed = all_passed
ledger.save_run(
workspace,
ProofRun(
Expand Down
15 changes: 15 additions & 0 deletions codeframe/ui/routers/_helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
"""Shared helpers for v2 routers."""

import json
import os
from pathlib import Path


def atomic_write_json(path: Path, payload: dict) -> None:
"""Write JSON via temp-file + os.replace so a crash mid-write cannot
leave a truncated file at `path`.
"""
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(payload, indent=2))
os.replace(tmp, path)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
97 changes: 90 additions & 7 deletions codeframe/ui/routers/proof_v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,15 @@
GET /api/v2/proof/requirements/{req_id}/evidence list_evidence()
"""

import json
import logging
import time
import uuid
from datetime import date
from typing import Any, Optional
from typing import Any, Literal, Optional

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator

from codeframe.core.proof.capture import capture_requirement
from codeframe.core.proof.ledger import (
Expand All @@ -33,6 +34,8 @@
waive_requirement,
)
from codeframe.core.proof.models import (
PROOF9_GATE_ORDER,
PROOF_CONFIG_FILENAME,
Gate,
ReqStatus,
Severity,
Expand All @@ -44,6 +47,7 @@
from codeframe.lib.rate_limiter import rate_limit_ai, rate_limit_standard
from codeframe.ui.dependencies import get_v2_workspace
from codeframe.ui.response_models import ErrorCodes, api_error
from codeframe.ui.routers._helpers import atomic_write_json

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -413,11 +417,18 @@ async def run_proof_endpoint(
for req_id, gate_results in results.items()
}

passed = all(
satisfied
for gate_results in results.values()
for _, satisfied in gate_results
)
# Use the strictness-aware overall_passed that run_proof() persisted,
# so warn-mode does not surface as failure via the cached run status.
persisted_run = get_run(workspace, run_id)
if persisted_run is not None:
passed = persisted_run.overall_passed
else:
# Fallback only if persistence unexpectedly failed
passed = all(
satisfied
for gate_results in results.values()
for _, satisfied in gate_results
)
response = RunProofResponse(
success=True,
run_id=run_id,
Expand Down Expand Up @@ -634,6 +645,78 @@ async def get_run_evidence_endpoint(
)


# ============================================================================
# PROOF9 Config (issue #556)
#
# Persists which gates are enabled by default and the strictness setting
# (strict vs warn) to .codeframe/proof_config.json.
# ============================================================================


_VALID_GATES = {g.value for g in Gate}


def _proof_config_path(workspace: Workspace):
return workspace.state_dir / PROOF_CONFIG_FILENAME


def _default_proof_config() -> dict:
return {"enabled_gates": list(PROOF9_GATE_ORDER), "strictness": "strict"}


class ProofConfigResponse(BaseModel):
enabled_gates: list[str]
strictness: Literal["strict", "warn"]


class UpdateProofConfigRequest(BaseModel):
enabled_gates: list[str]
strictness: Literal["strict", "warn"]

@field_validator("enabled_gates")
@classmethod
def _validate_gates(cls, v: list[str]) -> list[str]:
unknown = [g for g in v if g not in _VALID_GATES]
if unknown:
raise ValueError(
f"Unknown gate(s): {unknown}. Valid: {sorted(_VALID_GATES)}"
)
return v


@router.get("/config", response_model=ProofConfigResponse)
@rate_limit_standard()
async def get_proof_config(
request: Request,
workspace: Workspace = Depends(get_v2_workspace),
) -> ProofConfigResponse:
"""Load PROOF9 defaults for this workspace.

Returns the all-gates-enabled + strict defaults if no config file exists.
"""
path = _proof_config_path(workspace)
if path.exists():
try:
data = json.loads(path.read_text())
return ProofConfigResponse(**data)
except (OSError, json.JSONDecodeError, ValueError) as e:
logger.warning("Invalid proof_config.json — falling back to defaults: %s", e)
return ProofConfigResponse(**_default_proof_config())


@router.put("/config", response_model=ProofConfigResponse)
@rate_limit_standard()
async def update_proof_config(
request: Request,
body: UpdateProofConfigRequest,
workspace: Workspace = Depends(get_v2_workspace),
) -> ProofConfigResponse:
"""Persist PROOF9 defaults to .codeframe/proof_config.json."""
payload = {"enabled_gates": body.enabled_gates, "strictness": body.strictness}
atomic_write_json(_proof_config_path(workspace), payload)
return ProofConfigResponse(**payload)
Comment on lines +650 to +721
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major | 🏗️ Heavy lift

Move PROOF config policy out of the router.

This block now owns default values, gate validation, file layout, and corrupted-file fallback behavior. That is domain logic, and keeping it in codeframe/ui/routers/proof_v2.py creates another place that can drift from the core runner’s interpretation of the same config. Please move load/save/default/validate behavior into a core API and keep the router to request/response translation.

As per coding guidelines, codeframe/ui/**/*.py: "FastAPI server and web UI are thin adapters over core — do NOT implement domain logic in codeframe/ui/ routers".

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@codeframe/ui/routers/proof_v2.py` around lines 648 - 717, The router
currently contains domain logic for defaults, validation, file layout, and
corrupted-file fallback (symbols: _VALID_GATES, _proof_config_path,
_default_proof_config, UpdateProofConfigRequest._validate_gates,
get_proof_config, update_proof_config, atomic_write_json); move that logic into
a core API (e.g., core.proof_config) by implementing functions like
default_proof_config(), validate_gates(gates) or exceptions for invalid gates,
load_proof_config(workspace) which reads the file, returns defaults on
missing/corrupt files and logs or surfaces errors, and
save_proof_config(workspace, payload) which does atomic write; then simplify the
router handlers to only call core.load_proof_config and core.save_proof_config
and map request/response models (keep ProofConfigResponse and
UpdateProofConfigRequest as DTOs) so all file/validation behavior lives in the
core module instead of codeframe/ui/routers/proof_v2.py.



@router.get("/requirements/{req_id}/evidence", response_model=list[EvidenceResponse])
@rate_limit_standard()
async def list_evidence_endpoint(
Expand Down
87 changes: 87 additions & 0 deletions codeframe/ui/routers/workspace_v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
PATCH /api/v2/workspaces/current - Update workspace (e.g., tech stack)
"""

import json
import logging
from pathlib import Path
from typing import Optional
Expand All @@ -21,6 +22,7 @@
from codeframe.ui.dependencies import get_v2_workspace
from codeframe.core.workspace import Workspace
from codeframe.ui.response_models import api_error, ErrorCodes
from codeframe.ui.routers._helpers import atomic_write_json

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -260,6 +262,91 @@ async def update_current_workspace(
)


# ============================================================================
# Workspace Config (issue #556)
#
# Persists per-workspace UI-configurable settings (root path display, default
# branch, tech-stack auto-detect toggle + manual override) to
# .codeframe/workspace_config.json.
# ============================================================================


_WORKSPACE_CONFIG_FILENAME = "workspace_config.json"


class WorkspaceConfigResponse(BaseModel):
workspace_root: str = Field(
..., description="Display-only. The server resolves the active workspace from the workspace_path query parameter."
)
default_branch: str
auto_detect_tech_stack: bool
tech_stack_override: Optional[str] = None


class UpdateWorkspaceConfigRequest(BaseModel):
workspace_root: str = Field(
...,
min_length=1,
description="Display-only — editing does not relocate the workspace.",
)
default_branch: str = Field(..., min_length=1)
auto_detect_tech_stack: bool
tech_stack_override: Optional[str] = None


def _workspace_config_path(workspace: Workspace) -> Path:
return workspace.state_dir / _WORKSPACE_CONFIG_FILENAME


def _default_workspace_config(workspace: Workspace) -> dict:
return {
"workspace_root": str(workspace.repo_path),
"default_branch": "main",
"auto_detect_tech_stack": True,
"tech_stack_override": None,
}


@router.get("/config", response_model=WorkspaceConfigResponse)
@rate_limit_standard()
async def get_workspace_config(
request: Request,
workspace: Workspace = Depends(get_v2_workspace),
) -> WorkspaceConfigResponse:
"""Load workspace configuration for this workspace.

Returns defaults sourced from the Workspace itself if no config file exists.
"""
path = _workspace_config_path(workspace)
if path.exists():
try:
data = json.loads(path.read_text())
return WorkspaceConfigResponse(**data)
except (OSError, json.JSONDecodeError, ValueError) as e:
logger.warning(
"Invalid workspace_config.json — falling back to defaults: %s", e
)
return WorkspaceConfigResponse(**_default_workspace_config(workspace))


@router.put("/config", response_model=WorkspaceConfigResponse)
@rate_limit_standard()
async def update_workspace_config(
request: Request,
body: UpdateWorkspaceConfigRequest,
workspace: Workspace = Depends(get_v2_workspace),
) -> WorkspaceConfigResponse:
"""Persist workspace configuration to .codeframe/workspace_config.json.

Note: `workspace_root` is informational/display-only. The server resolves
the active workspace path from the `workspace_path` query parameter or
its default — editing this field does not relocate the workspace.
"""
payload = body.model_dump()
atomic_write_json(_workspace_config_path(workspace), payload)
return WorkspaceConfigResponse(**payload)


@router.get("/exists")
@rate_limit_standard()
async def check_workspace_exists(
Expand Down
Loading
Loading