diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md
index a4a935e..39f6dc7 100644
--- a/docs/implementation-plan.md
+++ b/docs/implementation-plan.md
@@ -27,6 +27,12 @@ Inference now splits three ways:
- searched lattice for the full local workflow
- direct autocorr lattice for the hosted demo path: one consensus size, one cheap phase probe, no size search, no phase rerank
+The hosted web shell now also has an auth seam:
+
+- `src/repixelizer/access.py` owns the app-local access boundary and local ownership checks
+- `src/repixelizer/gui.py` owns hosted route policy, landing-page behavior, and queue/job stamping
+- self-host and local runs stay permissive by default unless `GC_ACCESS_*` turns the seam on
+
## What is working
- lattice size and phase inference are still shared and still CUDA-capable
@@ -121,6 +127,15 @@ Rules:
- if a new term lands, it must belong cleanly in the one-field objective
- if a result looks better and the metrics disagree, fix the metrics rather than worshipping them
+### 4. Keep hosted auth surgery out of the solver
+
+Current stance:
+
+- Heimdall integration belongs in the hosted web layer, not in the reconstruction pipeline
+- queued jobs now have local owner metadata so future auth can resolve job access from local session/account instead of provider ids
+- the repo is only "auth-ready", not "auth-landed"
+- `GC_ACCESS_MODE=trusted-header` exists as a thin local seam for future integration and route-policy tests; it is not the final Heimdall story
+
## Guardrails
- prefer one clear hypothesis per pass
diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md
index 9e4c32b..ae94210 100644
--- a/notes/fresh-workspace-handoff.md
+++ b/notes/fresh-workspace-handoff.md
@@ -30,6 +30,7 @@ Do not trust this file for the exact live HEAD. Always check git.
- The live pipeline is `source -> lattice inference -> edge analysis -> phase-field reconstruction -> cleanup -> optional palette fit -> diagnostics`.
- searched lattice inference now trusts autocorrelation only; it keeps a tiny near-best lag plateau and looks for cross-axis consensus before collapsing to one lattice family. the old change-interval / pixel-walk sizing path was cut after it kept steering the badge away from the `126` family, and the projected edge-energy spectral pass was also removed after floor-hugging to 2px / 626-family aliases on the dense landscape fixture
- hosted demo inference now uses one direct autocorr-consensus size plus the cheap inferred phase, capped by the hosted output limit. it no longer silently pins unspecified jobs to `max_output_dimension`
+- hosted web access now has an explicit seam in `src/repixelizer/access.py`; `src/repixelizer/gui.py` binds request subjects, guards hosted routes, and stamps queued jobs with local owner metadata for future Heimdall wiring
- The low-confidence phase rerank path is a short preview solve, not a second full optimizer.
- the phase field now has a wider displacement budget, so moderate phase mistakes are allowed to get corrected by the field instead of relying purely on rerank ceremony
- Cleanup is secondary and usually a no-op; the core result is supposed to come from the solver, not cleanup cosplay.
@@ -56,6 +57,7 @@ The current spine:
- lattice inference with fixed-size and searched-size paths
- searched lattice inference currently trusts autocorrelation only, not the rejected pixel-walk or spectral priors
- hosted demo inference uses the direct autocorr path and keeps phase rerank off
+- hosted route auth belongs in `src/repixelizer/access.py` + `src/repixelizer/gui.py`, not in the solver or pipeline
- low-confidence phase rerank through short preview reconstruction
- edge analysis feeding one projected displacement-field optimizer with a wider displacement leash
- nearest-source final sampling from `uv0_px + disp_t`
@@ -74,6 +76,7 @@ The exact current control flow is documented in
- Do not let `state/evidence.jsonl` turn into an activity feed.
- Do not restart broad exploratory surgery when the current weak spot is still one bounded tapered-contour blemish.
- Do not drift into desktop-app fantasies just because the web GUI is nice now. The default commercial path is still hosted access.
+- Do not let future Heimdall work smear auth logic into `src/repixelizer/pipeline.py`, `src/repixelizer/inference.py`, or the solver stack. The seam is already in the hosted web layer; use it.
## Verification Guardrails
diff --git a/src/repixelizer/access.py b/src/repixelizer/access.py
new file mode 100644
index 0000000..426b559
--- /dev/null
+++ b/src/repixelizer/access.py
@@ -0,0 +1,231 @@
+from __future__ import annotations
+
+import contextlib
+import contextvars
+import os
+from dataclasses import dataclass, field
+from typing import Any, Iterator
+
+
+_CURRENT_ACCESS_SUBJECT: contextvars.ContextVar["AccessSubject | None"] = contextvars.ContextVar(
+ "repixelizer_current_access_subject",
+ default=None,
+)
+
+_PERMISSIVE_CAPABILITIES = frozenset(
+ {
+ "app_access",
+ "queue_submit",
+ "job_read_own",
+ "job_cancel_own",
+ "admin_access",
+ }
+)
+
+
+def _env_text(name: str) -> str | None:
+ raw = os.getenv(name)
+ if raw is None:
+ return None
+ text = raw.strip()
+ return text or None
+
+
+def _env_flag(name: str, default: bool) -> bool:
+ raw = os.getenv(name)
+ if raw is None:
+ return default
+ return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _parse_capabilities(raw: str | None) -> frozenset[str]:
+ if not raw:
+ return frozenset()
+ return frozenset(part.strip() for part in raw.split(",") if part.strip())
+
+
+@dataclass(frozen=True, slots=True)
+class AccessSubject:
+ account_id: str | None = None
+ session_id: str | None = None
+ access_revision: int | None = None
+ capabilities: frozenset[str] = field(default_factory=frozenset)
+ display_name: str | None = None
+ auth_mode: str = "off"
+ claims: dict[str, Any] = field(default_factory=dict)
+
+ @property
+ def is_authenticated(self) -> bool:
+ return self.account_id is not None or self.session_id is not None
+
+ def has_capability(self, capability: str) -> bool:
+ return capability in self.capabilities
+
+ def owns_job(self, *, account_id: str | None, session_id: str | None) -> bool:
+ if account_id is not None and self.account_id is not None and account_id == self.account_id:
+ return True
+ if session_id is not None and self.session_id is not None and session_id == self.session_id:
+ return True
+ return False
+
+ def to_public_json(self) -> dict[str, Any]:
+ return {
+ "authenticated": self.is_authenticated,
+ "accountId": self.account_id,
+ "sessionId": self.session_id,
+ "accessRevision": self.access_revision,
+ "displayName": self.display_name,
+ "capabilities": sorted(self.capabilities),
+ "authMode": self.auth_mode,
+ }
+
+
+@dataclass(frozen=True, slots=True)
+class AccessRuntimeConfig:
+ mode: str
+ app_slug: str
+ required: bool
+ protect_queue: bool
+ login_url: str | None
+ logout_url: str | None
+
+ @property
+ def enabled(self) -> bool:
+ return self.mode != "off"
+
+ def to_public_json(self) -> dict[str, Any]:
+ return {
+ "enabled": self.enabled,
+ "mode": self.mode,
+ "required": self.required,
+ "protectQueue": self.protect_queue,
+ "appSlug": self.app_slug,
+ "loginUrl": self.login_url,
+ "logoutUrl": self.logout_url,
+ }
+
+
+class AccessDenied(RuntimeError):
+ def __init__(self, status_code: int, detail: str) -> None:
+ super().__init__(detail)
+ self.status_code = int(status_code)
+ self.detail = detail
+
+
+@contextlib.contextmanager
+def bind_access_subject(subject: AccessSubject) -> Iterator[None]:
+ token = _CURRENT_ACCESS_SUBJECT.set(subject)
+ try:
+ yield
+ finally:
+ _CURRENT_ACCESS_SUBJECT.reset(token)
+
+
+class AccessController:
+ def __init__(self, runtime: AccessRuntimeConfig) -> None:
+ self.runtime = runtime
+
+ @classmethod
+ def from_env(cls, *, hosted_demo: bool) -> "AccessController":
+ mode = (_env_text("GC_ACCESS_MODE") or "off").strip().lower()
+ if mode not in {"off", "trusted-header"}:
+ mode = "off"
+ required_default = hosted_demo if mode != "off" else False
+ runtime = AccessRuntimeConfig(
+ mode=mode,
+ app_slug=_env_text("GC_ACCESS_APP_SLUG") or "repixelizer",
+ required=_env_flag("GC_ACCESS_REQUIRED", required_default),
+ protect_queue=_env_flag("GC_ACCESS_PROTECT_QUEUE", False),
+ login_url=_env_text("GC_ACCESS_LOGIN_URL"),
+ logout_url=_env_text("GC_ACCESS_LOGOUT_URL"),
+ )
+ return cls(runtime)
+
+ def public_payload(self) -> dict[str, Any]:
+ return self.runtime.to_public_json()
+
+ def current_subject(self) -> AccessSubject:
+ subject = _CURRENT_ACCESS_SUBJECT.get()
+ if subject is not None:
+ return subject
+ if self.runtime.mode == "off":
+ return AccessSubject(
+ capabilities=_PERMISSIVE_CAPABILITIES,
+ auth_mode=self.runtime.mode,
+ )
+ return AccessSubject(auth_mode=self.runtime.mode)
+
+ def peek_request_subject(self, request: Any) -> AccessSubject:
+ return self._subject_from_request(request, allow_anonymous=True)
+
+ def require_request_capability(self, request: Any, capability: str) -> AccessSubject:
+ subject = self._subject_from_request(request, allow_anonymous=not self.runtime.required)
+ self._require_capability(subject, capability)
+ return subject
+
+ def require_current_capability(self, capability: str) -> AccessSubject:
+ subject = self.current_subject()
+ self._require_capability(subject, capability)
+ return subject
+
+ def require_current_job_access(self, job: Any, capability: str) -> AccessSubject:
+ subject = self.require_current_capability(capability)
+ self.require_subject_job_access(subject, job)
+ return subject
+
+ def require_subject_job_access(self, subject: AccessSubject, job: Any) -> None:
+ if self.runtime.mode == "off":
+ return
+ account_id = getattr(job, "account_id", None)
+ session_id = getattr(job, "session_id", None)
+ if account_id is None and session_id is None:
+ raise AccessDenied(403, "Job has no local ownership metadata to validate against.")
+ if not subject.owns_job(account_id=account_id, session_id=session_id):
+ raise AccessDenied(403, "That job belongs to a different local account or session.")
+
+ def app_gate_redirect_url(self) -> str:
+ return self.runtime.login_url or "/"
+
+ def _require_capability(self, subject: AccessSubject, capability: str) -> None:
+ if self.runtime.mode == "off":
+ return
+ if not subject.is_authenticated and self.runtime.required:
+ raise AccessDenied(401, "Sign-in required.")
+ if not subject.has_capability(capability):
+ raise AccessDenied(403, f"Missing required capability '{capability}'.")
+
+ def _subject_from_request(self, request: Any, *, allow_anonymous: bool) -> AccessSubject:
+ if self.runtime.mode == "off":
+ return AccessSubject(
+ capabilities=_PERMISSIVE_CAPABILITIES,
+ auth_mode=self.runtime.mode,
+ )
+ if self.runtime.mode == "trusted-header":
+ return self._trusted_header_subject(request, allow_anonymous=allow_anonymous)
+ return AccessSubject(auth_mode=self.runtime.mode)
+
+ def _trusted_header_subject(self, request: Any, *, allow_anonymous: bool) -> AccessSubject:
+ headers = request.headers
+ account_id = headers.get("x-gc-account-id")
+ session_id = headers.get("x-gc-session-id")
+ access_revision_raw = headers.get("x-gc-access-revision")
+ display_name = headers.get("x-gc-display-name")
+ capabilities = _parse_capabilities(headers.get("x-gc-capabilities"))
+ access_revision: int | None = None
+ if access_revision_raw is not None:
+ try:
+ access_revision = int(access_revision_raw)
+ except ValueError:
+ access_revision = None
+ if account_id is None and session_id is None:
+ if allow_anonymous:
+ return AccessSubject(auth_mode=self.runtime.mode)
+ raise AccessDenied(401, "Sign-in required.")
+ return AccessSubject(
+ account_id=account_id,
+ session_id=session_id,
+ access_revision=access_revision,
+ capabilities=capabilities,
+ display_name=display_name,
+ auth_mode=self.runtime.mode,
+ )
diff --git a/src/repixelizer/gui.py b/src/repixelizer/gui.py
index 7bc527b..3319c06 100644
--- a/src/repixelizer/gui.py
+++ b/src/repixelizer/gui.py
@@ -16,6 +16,7 @@ from typing import Any
import numpy as np
from PIL import Image, ImageDraw
+from .access import AccessController, AccessDenied, bind_access_subject
from .diagnostics import _displacement_preview_rgba
from .inference import inference_to_json
from .observe import PipelineCancelled
@@ -237,6 +238,9 @@ class GuiJob:
filename: str
options: dict[str, Any]
spool_path: Path
+ account_id: str | None = None
+ session_id: str | None = None
+ access_revision: int | None = None
created_at: float = field(default_factory=time.time)
last_heartbeat_at: float = field(default_factory=time.time)
phase_field_preview_stride: int = 4
@@ -648,7 +652,16 @@ class GuiJobManager:
self._started = False
self._purge_spool_dir()
- def submit_job(self, *, filename: str, raw: bytes, options: dict[str, Any]) -> GuiJob:
+ def submit_job(
+ self,
+ *,
+ filename: str,
+ raw: bytes,
+ options: dict[str, Any],
+ account_id: str | None = None,
+ session_id: str | None = None,
+ access_revision: int | None = None,
+ ) -> GuiJob:
self.start()
suffix = Path(filename or "input.png").suffix or ".png"
job = GuiJob(
@@ -656,6 +669,9 @@ class GuiJobManager:
filename=filename or "input.png",
options=options,
spool_path=self.config.spool_dir / f"{uuid.uuid4().hex}{suffix}",
+ account_id=account_id,
+ session_id=session_id,
+ access_revision=access_revision,
phase_field_preview_stride=self.config.phase_field_preview_stride,
)
job.spool_path.write_bytes(raw)
@@ -861,12 +877,27 @@ def _versioned_gui_index(static_dir: Path) -> str:
return html
-def _landing_page_html(config: HostedDemoConfig) -> str:
+def _landing_page_html(config: HostedDemoConfig, auth: dict[str, Any] | None = None) -> str:
demo_limits = (
f"{config.max_output_dimension}px output cap, "
f"{config.default_steps} default solver steps, "
f"{config.queue_capacity} waiting slots"
)
+ auth_enabled = bool(auth and auth.get("enabled"))
+ login_url = None if auth is None else auth.get("loginUrl")
+ primary_href = login_url or "/app/"
+ primary_label = "Sign in to the demo" if auth_enabled else "Open the demo"
+ hero_note = (
+ "Hosted access will be gated through Heimdall once the auth pass lands."
+ if auth_enabled
+ else "It is not a miracle worker, and it is definitely not here to pretend it invented hand-authored pixel art from nothing."
+ )
+ auth_notice = ""
+ if auth_enabled and not login_url:
+ auth_notice = (
+ '
Auth is enabled here, but no Heimdall login URL has been wired yet. '
+ 'That means the front door is ready and the key is still missing.
'
+ )
return f"""
@@ -1091,7 +1122,7 @@ def _landing_page_html(config: HostedDemoConfig) -> str:
Repixelizer takes AI-slop pixel cosplay, infers the lattice it should have had, then drags it back onto an actual coherent mosaic.
Self-hosters can still go straight to the app. Hosted mode just gets a front door instead of making everyone trip over raw controls immediately.
+ {auth_notice}
@@ -1155,6 +1187,7 @@ def create_app():
config = HostedDemoConfig.from_env()
manager = GuiJobManager(config)
+ access_controller = AccessController.from_env(hosted_demo=config.hosted_demo)
@contextlib.asynccontextmanager
async def lifespan(_app):
@@ -1176,13 +1209,71 @@ def create_app():
response.headers["Expires"] = "0"
return response
+ @app.middleware("http")
+ async def bind_access_context(request, call_next):
+ path = request.url.path
+
+ async def call_with_subject(subject):
+ with bind_access_subject(subject):
+ return await call_next(request)
+
+ def resolve_job_subject(capability: str):
+ parts = [part for part in path.split("/") if part]
+ if len(parts) < 3:
+ return access_controller.peek_request_subject(request)
+ job = manager.get_job(parts[2])
+ if job is None:
+ return access_controller.peek_request_subject(request)
+ subject = access_controller.require_request_capability(request, capability)
+ access_controller.require_subject_job_access(subject, job)
+ return subject
+
+ try:
+ if path in {"/", "/api/health", "/api/config", "/api/auth/session"}:
+ subject = access_controller.peek_request_subject(request)
+ elif path == "/api/queue":
+ if access_controller.runtime.protect_queue:
+ subject = access_controller.require_request_capability(request, "app_access")
+ else:
+ subject = access_controller.peek_request_subject(request)
+ elif path == "/api/jobs" and request.method.upper() == "POST":
+ subject = access_controller.require_request_capability(request, "queue_submit")
+ elif path.startswith("/api/jobs/"):
+ capability = "job_cancel_own" if request.method.upper() == "DELETE" else "job_read_own"
+ subject = resolve_job_subject(capability)
+ elif path == "/app" or path.startswith("/app/"):
+ if access_controller.runtime.enabled and access_controller.runtime.required:
+ subject = access_controller.require_request_capability(request, "app_access")
+ else:
+ subject = access_controller.peek_request_subject(request)
+ else:
+ subject = access_controller.peek_request_subject(request)
+ except AccessDenied as exc:
+ if path == "/app" or path.startswith("/app/"):
+ return RedirectResponse(url=access_controller.app_gate_redirect_url(), status_code=303)
+ return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)
+
+ return await call_with_subject(subject)
+
@app.get("/api/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.get("/api/config")
def get_config():
- return JSONResponse(config.public_payload())
+ payload = config.public_payload()
+ payload["auth"] = access_controller.public_payload()
+ return JSONResponse(payload)
+
+ @app.get("/api/auth/session")
+ def get_auth_session():
+ subject = access_controller.current_subject()
+ return JSONResponse(
+ {
+ "auth": access_controller.public_payload(),
+ "subject": subject.to_public_json(),
+ }
+ )
@app.get("/api/queue")
def get_queue():
@@ -1190,9 +1281,15 @@ def create_app():
@app.get("/api/jobs/{job_id}")
def get_job(job_id: str):
- payload = manager.get_job_state_payload(job_id)
- if payload is None:
+ job = manager.get_job(job_id)
+ if job is None:
raise HTTPException(status_code=404, detail="Unknown job")
+ try:
+ access_controller.require_current_job_access(job, "job_read_own")
+ except AccessDenied as exc:
+ raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
+ payload = manager.get_job_state_payload(job_id)
+ assert payload is not None
return JSONResponse(payload)
@app.post("/api/jobs")
@@ -1227,7 +1324,17 @@ def create_app():
strip_background=strip_background,
skip_phase_rerank=skip_phase_rerank,
)
- job = manager.submit_job(filename=filename, raw=raw, options=options)
+ subject = access_controller.require_current_capability("queue_submit")
+ job = manager.submit_job(
+ filename=filename,
+ raw=raw,
+ options=options,
+ account_id=subject.account_id,
+ session_id=subject.session_id,
+ access_revision=subject.access_revision,
+ )
+ except AccessDenied as exc:
+ raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
except QueueFullError as exc:
raise HTTPException(status_code=429, detail=str(exc)) from exc
except ValueError as exc:
@@ -1253,16 +1360,28 @@ def create_app():
@app.post("/api/jobs/{job_id}/heartbeat")
def job_heartbeat(job_id: str):
- payload = manager.heartbeat(job_id)
- if payload is None:
+ job = manager.get_job(job_id)
+ if job is None:
raise HTTPException(status_code=404, detail="Unknown job")
+ try:
+ access_controller.require_current_job_access(job, "job_read_own")
+ except AccessDenied as exc:
+ raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
+ payload = manager.heartbeat(job_id)
+ assert payload is not None
return JSONResponse(payload)
@app.delete("/api/jobs/{job_id}")
def cancel_job(job_id: str):
- payload = manager.cancel_job(job_id, "Canceled because the browser left or explicitly bailed.")
- if payload is None:
+ job = manager.get_job(job_id)
+ if job is None:
raise HTTPException(status_code=404, detail="Unknown job")
+ try:
+ access_controller.require_current_job_access(job, "job_cancel_own")
+ except AccessDenied as exc:
+ raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
+ payload = manager.cancel_job(job_id, "Canceled because the browser left or explicitly bailed.")
+ assert payload is not None
return JSONResponse(payload)
@app.get("/api/jobs/{job_id}/events")
@@ -1270,6 +1389,10 @@ def create_app():
job = manager.get_job(job_id)
if job is None:
raise HTTPException(status_code=404, detail="Unknown job")
+ try:
+ access_controller.require_current_job_access(job, "job_read_own")
+ except AccessDenied as exc:
+ raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
def generate():
index = 0
@@ -1306,7 +1429,7 @@ def create_app():
@app.get("/")
def root():
if config.hosted_demo:
- return HTMLResponse(_landing_page_html(config))
+ return HTMLResponse(_landing_page_html(config, access_controller.public_payload()))
return RedirectResponse(url="/app/")
else: # pragma: no cover - only relevant before frontend assets exist
@app.get("/")
diff --git a/state/evidence.jsonl b/state/evidence.jsonl
index 82e9067..cbfea89 100644
--- a/state/evidence.jsonl
+++ b/state/evidence.jsonl
@@ -12,3 +12,4 @@
{"ts": "2026-04-26T17:58:43+00:00", "type": "research", "status": "ok", "note": "Autocorr inference now keeps a tiny near-best lag plateau and searches for cross-axis consensus before collapsing to one lattice size. On the real badge fixture this preserves x-axis support for 10px alongside 11px, raises the shared prior to the 125-family, and restores auto inference to 125 with top candidates 125/126/124/127/123 while dense landscape remains 418x418."}
{"ts": "2026-04-26T18:19:52+00:00", "type": "design", "status": "accepted", "note": "Hosted demo jobs now use direct autocorr lattice inference with one cheap phase probe, capped by the hosted output limit. The hosted path no longer silently pins unspecified jobs to max_output_dimension, and phase rerank is forced off there while the full local workflow keeps searched autocorr inference available."}
{"ts": "2026-04-26T22:40:02+00:00", "type": "design", "status": "accepted", "note": "Hosted mode now owns a distinct public front door in src/repixelizer/gui.py: / serves a landing page there, while local and self-host runs still redirect root straight to /app/. This keeps future auth and member-gating surgery in the hosted web layer instead of smearing it into the self-deploy path or the solver stack."}
+{"ts": "2026-04-26T22:57:43+00:00", "type": "design", "status": "accepted", "note": "Repixelizer is now auth-ready at the hosted web boundary without landing real Heimdall auth. src/repixelizer/access.py owns the local access seam, src/repixelizer/gui.py binds request subjects and guards hosted routes, queued jobs now carry local owner metadata, and self-host remains permissive by default unless GC_ACCESS_* turns the seam on."}
diff --git a/state/map.yaml b/state/map.yaml
index d7eea4d..dc9532a 100644
--- a/state/map.yaml
+++ b/state/map.yaml
@@ -58,6 +58,7 @@ active_subgoals:
- Improve along-stroke versus across-stroke behavior near tapered contours in a way that still fits the one-field optimizer model.
- Keep inference, compare mode, diagnostics, tuning, and GUI observer events working after solver changes.
- Preserve `source_structure` as a real objective signal instead of over-trusting lattice-only fidelity.
+ - Keep hosted auth integration isolated to the web layer so Heimdall can wire it in later without contaminating self-host or solver code.
accepted_design:
- `phase-field` in `src/repixelizer/phase_field.py` is the only live reconstruction engine.
@@ -67,6 +68,7 @@ accepted_design:
- `src/repixelizer/phase_field.py` owns the one honest optimization variable: `disp_t`.
- Hosted demo jobs now use a direct autocorr lattice path: one consensus size, one cheap phase probe, capped by the hosted output limit, with phase rerank disabled.
- Hosted mode owns the public front door in `src/repixelizer/gui.py`: `/` is a landing page there, while local/self-host runs keep root as a plain redirect to `/app/`.
+ - `src/repixelizer/access.py` now owns the local access seam for hosted routes, and queued jobs carry `account_id`, `session_id`, and `access_revision` for future Heimdall-backed ownership checks.
- The optimizer starts from fixed lattice centers, uses a wider displacement budget than the earlier tighter leash, optimizes displacement with projected Adam, then rounds to real source pixels for final output.
- Cleanup is local and secondary; it does not define the core reconstruction model.
- `source_fidelity` is useful but incomplete; `source_structure` exists because visible structure can improve while lattice-only scoring complains.
@@ -93,5 +95,5 @@ rejected_paths:
reason: the projected edge-energy spectral pass floor-hugged to 2px / 626-family aliases on the dense landscape fixture and did not justify its complexity
current_status:
- summary: The repo has one live reconstruction engine, `phase-field`, wired through the main pipeline with lattice inference, edge analysis, projected displacement optimization, cleanup, optional palette fitting, diagnostics, compare mode, tuning, and GUI observer events. The searched lattice path now trusts autocorrelation only; it keeps a tiny near-best lag plateau and uses cross-axis consensus before collapsing to one lattice family. The hosted demo path now skips multi-size lattice search and phase rerank entirely, taking one direct autocorr-consensus size plus the cheap inferred phase and respecting the hosted output cap instead of silently pinning unspecified jobs to that cap. The spectral prior experiment was removed after it floor-hugged to 2px / 626-family aliases on the dense landscape fixture, and the old change-interval / pixel-walk sizing path was cut after it kept pruning the badge away from the `126` family. The solver now has a wider displacement budget so moderate phase error can be recovered by the field instead of requiring as much rerank ceremony. `source_structure` now rides alongside `source_fidelity` because lattice-only scoring was lying about visible quality. The tracked weakness is still the widened dark contour near the badge sword tip on tapered shapes.
+ summary: The repo has one live reconstruction engine, `phase-field`, wired through the main pipeline with lattice inference, edge analysis, projected displacement optimization, cleanup, optional palette fitting, diagnostics, compare mode, tuning, and GUI observer events. The searched lattice path now trusts autocorrelation only; it keeps a tiny near-best lag plateau and uses cross-axis consensus before collapsing to one lattice family. The hosted demo path now skips multi-size lattice search and phase rerank entirely, taking one direct autocorr-consensus size plus the cheap inferred phase and respecting the hosted output cap instead of silently pinning unspecified jobs to that cap. Hosted web access now has its own seam in `src/repixelizer/access.py` and `src/repixelizer/gui.py`: route guards live there, self-host remains permissive by default, `/` stays a hosted-only landing page, and queued jobs now carry local owner metadata so Heimdall can wire real auth into the web layer later without dragging solver code into it. The spectral prior experiment was removed after it floor-hugged to 2px / 626-family aliases on the dense landscape fixture, and the old change-interval / pixel-walk sizing path was cut after it kept pruning the badge away from the `126` family. The solver now has a wider displacement budget so moderate phase error can be recovered by the field instead of requiring as much rerank ceremony. `source_structure` now rides alongside `source_fidelity` because lattice-only scoring was lying about visible quality. The tracked weakness is still the widened dark contour near the badge sword tip on tapered shapes.
next_action: After rehydrate, if the user wants continued solver work, take one bounded hypothesis aimed at along-stroke versus across-stroke behavior near tapered contours in `src/repixelizer/phase_field.py`, validate it against the pinned badge case and `tests/fixtures/real/ai-badge-tip-focus.json`, and revert immediately if the visual result or `source_structure` does not clearly improve.
diff --git a/tests/test_gui.py b/tests/test_gui.py
index dfbe8f0..4743287 100644
--- a/tests/test_gui.py
+++ b/tests/test_gui.py
@@ -10,10 +10,13 @@ from pathlib import Path
import numpy as np
from PIL import Image
+import pytest
from starlette.datastructures import UploadFile
+from fastapi import HTTPException
import repixelizer.inference as inference_module
import repixelizer.pipeline as pipeline_module
+from repixelizer.access import AccessSubject, bind_access_subject
from repixelizer.observe import PipelineCancelled
from repixelizer.pipeline import run_pipeline_rgba
from repixelizer.synthetic import fake_pixelize, make_emblem
@@ -21,25 +24,43 @@ from repixelizer.params import SolverHyperParams
from repixelizer.types import InferenceCandidate, InferenceResult, PhaseFieldSourceAnalysis, SolverArtifacts
-async def _get_response(app, path: str) -> tuple[int, dict[str, str], bytes]:
+async def _get_response(
+ app,
+ path: str,
+ *,
+ method: str = "GET",
+ headers: dict[str, str] | None = None,
+ body: bytes = b"",
+) -> tuple[int, dict[str, str], bytes]:
messages: list[dict[str, object]] = []
+ sent = False
async def receive():
- return {"type": "http.request", "body": b"", "more_body": False}
+ nonlocal sent
+ if sent:
+ return {"type": "http.disconnect"}
+ sent = True
+ return {"type": "http.request", "body": body, "more_body": False}
async def send(message):
messages.append(message)
+ raw_headers = []
+ if headers:
+ raw_headers.extend(
+ (key.lower().encode("latin-1"), value.encode("latin-1"))
+ for key, value in headers.items()
+ )
scope = {
"type": "http",
"asgi": {"version": "3.0"},
"http_version": "1.1",
- "method": "GET",
+ "method": method.upper(),
"scheme": "http",
"path": path,
"raw_path": path.encode("ascii"),
"query_string": b"",
- "headers": [],
+ "headers": raw_headers,
"client": ("127.0.0.1", 12345),
"server": ("testserver", 80),
"root_path": "",
@@ -323,6 +344,37 @@ def test_gui_hosted_root_serves_landing_page(monkeypatch, tmp_path: Path) -> Non
assert 'href="/app/"' in html
+def test_gui_hosted_app_redirects_to_landing_when_auth_required(monkeypatch, tmp_path: Path) -> None:
+ from repixelizer.gui import create_app
+
+ monkeypatch.setenv("REPIXELIZER_HOSTED_DEMO", "1")
+ monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(tmp_path / "spool"))
+ monkeypatch.setenv("GC_ACCESS_MODE", "trusted-header")
+ monkeypatch.setenv("GC_ACCESS_REQUIRED", "1")
+
+ app = create_app()
+ status, headers, _body = asyncio.run(_get_response(app, "/app/"))
+
+ assert status in {302, 303, 307}
+ assert headers["location"] == "/"
+
+
+def test_gui_api_jobs_denies_unauthenticated_request_when_auth_required(monkeypatch, tmp_path: Path) -> None:
+ from repixelizer.gui import create_app
+
+ monkeypatch.setenv("REPIXELIZER_HOSTED_DEMO", "1")
+ monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(tmp_path / "spool"))
+ monkeypatch.setenv("GC_ACCESS_MODE", "trusted-header")
+ monkeypatch.setenv("GC_ACCESS_REQUIRED", "1")
+
+ app = create_app()
+ status, _headers, body = asyncio.run(_get_response(app, "/api/jobs", method="POST"))
+ payload = _response_json(type("Response", (), {"body": body})())
+
+ assert status == 401
+ assert payload["detail"] == "Sign-in required."
+
+
def test_gui_local_root_keeps_redirect_to_app(monkeypatch, tmp_path: Path) -> None:
from repixelizer.gui import create_app
@@ -499,6 +551,26 @@ def test_gui_hosted_config_endpoint_exposes_demo_limits(monkeypatch, tmp_path: P
assert payload["ui"]["showDeviceControl"] is False
assert payload["ui"]["showStripBackgroundControl"] is False
assert payload["ui"]["showQueuePanel"] is True
+ assert payload["auth"]["enabled"] is False
+ assert payload["auth"]["mode"] == "off"
+
+
+def test_gui_auth_session_endpoint_reports_anonymous_subject_by_default(monkeypatch, tmp_path: Path) -> None:
+ from repixelizer.gui import create_app
+
+ monkeypatch.setenv("REPIXELIZER_HOSTED_DEMO", "1")
+ monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(tmp_path / "spool"))
+
+ app = create_app()
+ status, _headers, body = asyncio.run(_get_response(app, "/api/auth/session"))
+ payload = _response_json(type("Response", (), {"body": body})())
+
+ assert status == 200
+ assert payload["auth"]["enabled"] is False
+ assert payload["subject"]["authenticated"] is False
+ assert payload["subject"]["capabilities"] == sorted(
+ ["admin_access", "app_access", "job_cancel_own", "job_read_own", "queue_submit"]
+ )
def test_hosted_job_options_use_direct_autocorr_and_skip_rerank() -> None:
@@ -541,6 +613,62 @@ def test_hosted_job_options_use_direct_autocorr_and_skip_rerank() -> None:
assert options["skip_phase_rerank"] is True
+def test_gui_job_routes_enforce_bound_subject_ownership(monkeypatch, tmp_path: Path) -> None:
+ from repixelizer.gui import create_app
+
+ monkeypatch.setenv("REPIXELIZER_HOSTED_DEMO", "1")
+ monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(tmp_path / "spool"))
+ monkeypatch.setenv("GC_ACCESS_MODE", "trusted-header")
+ monkeypatch.setenv("GC_ACCESS_REQUIRED", "1")
+
+ app = create_app()
+ create_job = _route_endpoint(app, "/api/jobs", "POST")
+ get_job = _route_endpoint(app, "/api/jobs/{job_id}", "GET")
+
+ subject = AccessSubject(
+ account_id="acct-1",
+ session_id="sess-1",
+ access_revision=4,
+ capabilities=frozenset({"queue_submit", "job_read_own", "job_cancel_own"}),
+ auth_mode="trusted-header",
+ )
+ with bind_access_subject(subject):
+ created = asyncio.run(
+ create_job(
+ image=UploadFile(filename="tiny.png", file=io.BytesIO(_png_bytes())),
+ target_size=None,
+ target_width=None,
+ target_height=None,
+ phase_x=None,
+ phase_y=None,
+ steps=None,
+ seed=7,
+ device="auto",
+ strip_background=False,
+ skip_phase_rerank=False,
+ )
+ )
+ assert created.status_code == 200
+ job_id = _response_json(created)["jobId"]
+
+ with bind_access_subject(subject):
+ owned = get_job(job_id)
+ assert owned.status_code == 200
+
+ intruder = AccessSubject(
+ account_id="acct-2",
+ session_id="sess-2",
+ access_revision=8,
+ capabilities=frozenset({"job_read_own"}),
+ auth_mode="trusted-header",
+ )
+ with bind_access_subject(intruder):
+ with pytest.raises(HTTPException) as excinfo:
+ get_job(job_id)
+ assert excinfo.value.status_code == 403
+ assert "different local account or session" in str(excinfo.value.detail)
+
+
def test_gui_queue_panel_defaults_off_for_local_runs_and_can_be_forced(monkeypatch, tmp_path: Path) -> None:
from repixelizer.gui import create_app