GameCult/repixelizer
Watch
9
0
Fork
You've already forked repixelizer
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge the deployed Eve surface branch into main

Browse source 
Production has served codex/repixelizer-eve-surface at 01dfab9 since
2026-06-16. Idunn admits main (Q1 default, docs/idunn-deploy-cut.md Cut 1),
so main must carry what runs: the Eve surface, verse_state.py,
cultlib_support.py and the msgpack dependency.

main's only commit off the merge base, 9e8c8f6, removed bilinear_resize and
lattice_source_rgba. The branch never touched io.py or metrics.py and calls
neither function, so the removals survive the merge without conflict.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 12:37:35 +02:00
commit 8e4f6492d0
18 changed files with 3059 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

83
docs/eve-surface-lowering.md Normal file
View file

@ -0,0 +1,83 @@
# Repixelizer Eve Surface Lowering
Repixelizer now emits a read-only `gamecult.eve.surface.v1` projection for its
front page, single-page app workflow, runtime config, auth projection, queue
summary, and command intents.
## Authority Map
- Owner: `repixelizer.gui` still owns hosted config, route policy, queue state,
job ownership, uploads, solver execution, and browser assets.
- Inputs: `HostedDemoConfig`, `AccessController.public_payload()`, and
`GuiJobManager.get_queue_summary()`.
- Output: `repixelizer.eve_surface.build_repixelizer_eve_surface()` emits the
retained Eve surface tree, style tokens, command descriptors, and renderer
comparison notes.
- Derived state: `/eve/surface`, `repixelize eve-surface`, and
`docs/fixtures/repixelizer.eve-surface.json` are projections. They do not
decide queue truth.
- Forbidden writers: Eve renderers, Odin, Periwinkle, and fixtures must route
commands back through `/api/jobs`, `/api/jobs/{job_id}`, or `/app/`; they do
not mutate Repixelizer jobs directly.
- Shared paths: browser GUI, CLI fixture export, hosted `/eve/surface`, and
provider advertisement all name the same surface and route authority.
- Cut line: no parallel status-card service, no renderer-owned queue summary,
and no fake native canvas authority until Eve can lower upload and image-edit
controls coherently.
## Published Surface
Generate the deterministic fixture:
```powershell
repixelize eve-surface --out docs/fixtures/repixelizer.eve-surface.json --updated-at 2026-06-04T00:00:00Z --public-base-url https://repixelizer.gamecult.org
```
The hosted GUI serves the live projection at:
```text
/eve/surface
```
The provider advertisement now marks `repixelizer.operator.surface` as
`available` with an HTTP JSON lowering at `/eve/surface`.
## Retro Style Contract
The Eve style profile is `repixelizer.retro.pixel`. The tokens intentionally
mirror the browser CSS instead of flattening the app into generic dashboard
chrome:
- title font: `Press Start 2P`
- body/mono font: `VT323`
- dark shell gradient and blue panel surfaces
- warm yellow/orange accent tokens
- chunky pixel borders, lower border weight, corner accents, scanline overlay,
dashed inner frame, and pixelated image sampling
The browser lowering remains the visual oracle for this pass because it already
implements the full upload, solver inspection, comparison canvas, editor, and
retro CSS.
## Periwinkle Comparison
Periwinkle's Android Kotlin lowerer currently has useful structural support:
it connects to the Eve dashboard broker, traverses `surface.root`, renders
pane/card/text hierarchies, supports basic metric progress bars, and attaches
some command handlers.
The current gap is style and control parity. `MainActivity.kt` still hard-codes
`toneColor`, `textColorFor`, and `textSizeFor`; it does not consume
`surface.styles.tokens`, does not load Repixelizer's pixel fonts, does not
render image assets from the surface tree, and does not lower the upload picker,
comparison canvas, pan/zoom inspection, eyedropper, or paint controls.
Minimum Periwinkle parity work:
- map provider style tokens before applying fallback tone/text/font rules
- package or cache `Press Start 2P` and `VT323`
- render `assetUri` / `assetRef` as pixelated `ImageView` content
- lower Repixelizer command controls to Android document picker and job route
calls
- add native canvas controls only after ownership remains routed through the
existing Repixelizer job and editor command paths

1249
docs/fixtures/repixelizer.eve-surface.json Normal file
View file

File diff suppressed because it is too large Load diff

19
docs/fixtures/repixelizer.provider-advertisement.json
View file

@ -26,7 +26,7 @@
}
],
"freshness": {
"lastSeenAt": "2026-06-03T00:00:00Z",
"lastSeenAt": "2026-06-04T00:00:00Z",
"maxAgeMs": 15000,
"state": "fresh"
},
@ -139,6 +139,7 @@
],
"styleProfile": "repixelizer.product",
"tokenGroups": [
"repixelizer.retro.pixel",
"repixelizer.comparisonCanvas",
"repixelizer.cleanupTool",
"repixelizer.queuePanel",
@ -149,13 +150,19 @@
"surfaces": [
{
"audience": "operator",
"commands": [],
"commands": [
"repixelizer.web.open",
"repixelizer.job.submit",
"repixelizer.job.cancel_own"
],
"key": "cultmesh://repixelizer/surfaces/operator",
"mode": "read-only-first-cut",
"schema": "gamecult.eve.surface.v1",
"status": "planned",
"status": "available",
"styleProfile": "repixelizer.retro.pixel",
"surfaceId": "repixelizer.operator.surface",
"transport": "cultmesh-document"
"transport": "http-json",
"url": "https://repixelizer.gamecult.org/eve/surface"
},
{
"canonical": false,
@ -167,13 +174,13 @@
}
],
"title": "Repixelizer",
"updatedAt": "2026-06-03T00:00:00Z",
"updatedAt": "2026-06-04T00:00:00Z",
"verseId": "gamecult.local",
"witnesses": [
{
"freshness": {
"state": "planned",
"updatedAt": "2026-06-03T00:00:00Z"
"updatedAt": "2026-06-04T00:00:00Z"
},
"id": "repixelizer.hosted.witness",
"kind": "cc-export-path-reserved",

10
docs/repixelizer-witness-contract.md
View file

@ -67,14 +67,16 @@ The reserved witness path is `state/repixelizer.witness.cc`.
- schema catalog entries for the witness shapes above
- a reserved `.cc` witness entry with bulk image bytes, raw tokens, and private
claims marked as redacted
- a planned canonical Eve operator surface key:
- an available canonical Eve operator surface projection at `/eve/surface`,
retaining the canonical CultMesh surface key:
`cultmesh://repixelizer/surfaces/operator`
- the current browser lowering at `/app/`
- route-owned command boundaries for job submit and own-job cancellation
- route-owned command boundaries for web open, job submit, and own-job cancellation
- nested Verse declarations for session and hosted-operator spaces
- runtime projection from `HostedDemoConfig.from_env()` and
`AccessController.from_env()`
This is deliberately read-only. The current queue is still in-process memory.
The advertisement only tells discovery organs where truth lives and what shape
future durable witnesses must take.
The advertisement only tells discovery organs where truth lives, where the
read-only Eve projection is served, and what shape future durable witnesses must
take.

63
docs/verse-service-contract.md
View file

@ -59,6 +59,29 @@ The first pass should export these as a `.cc` witness from the current
single-process manager. A later pass can decide whether the typed store becomes
the primary queue owner.
The next live cut is not speculative anymore. The Python CultLib lane already
exists in `E:\Projects\CultLib\packages\cultcache-py`, including `cultcache_py`
for `.cc` persistence and `cultnet_py` for CultNet/RUDP transport. Repixelizer
needs to wire that runtime into the hosted GUI process so the live owner can:
- write a daemon-owned `.cc` witness from `GuiJobManager`
- publish daemon health to Idunn over CultNet/RUDP
- advertise command-boundary and transport-profile state from the runtime
- keep `/api/health`, `/api/config`, nginx, and systemd as compatibility
witnesses instead of the keepalive truth
## Current Compatibility Boundary
Today Yggdrasil still verifies Repixelizer through:
- `repixelizer-gui.service`
- `GET /api/health`
- `GET /api/config`
- nginx host-routed `/app/` and `/api/health`
Those checks prove deployment and public routing. They do not satisfy the daemon
truth contract Idunn expects.
## Eve Surface Target
Repixelizer should publish an Eve GUI/TUI DSL product surface with these panels
@ -87,7 +110,7 @@ tool palettes, and component variants. A Kotlin Android Eve runtime should be
able to run the cleanup flow with native controls while preserving the same
command boundary and product identity.
## Read-Only Witness Export
## Read-Only Witness And Eve Surface Export
The first read-only provider advertisement is generated by:
@ -96,9 +119,20 @@ repixelize witness-advertisement --out docs/fixtures/repixelizer.provider-advert
```
It names the witness schemas above, reserves `state/repixelizer.witness.cc` for
the future CultCache export, and publishes the planned Eve operator surface key
without changing queue ownership. The detailed authority map lives in
`docs/repixelizer-witness-contract.md`.
the future CultCache export, and publishes the available Eve operator surface
HTTP projection without changing queue ownership. The detailed authority maps
live in `docs/repixelizer-witness-contract.md` and
`docs/eve-surface-lowering.md`.
The read-only Eve surface fixture is generated by:
```powershell
repixelize eve-surface --out docs/fixtures/repixelizer.eve-surface.json
```
The hosted GUI serves the live projection at `/eve/surface`. It reads
`HostedDemoConfig`, `AccessController.public_payload()`, and
`GuiJobManager.get_queue_summary()`; it does not own queue truth.
## Migration Order
@ -107,15 +141,22 @@ without changing queue ownership. The detailed authority map lives in
2. Publish a read-only `gamecult.eve.provider_advertisement.v1` fixture/export
that names those shapes and the reserved `.cc` witness path.
3. Add a read-only export from the current `gui.py` queue/job manager into a
`.cc` witness.
4. Publish the witness through CultMesh.
5. Add an Eve DSL provider over the witness and existing health/config routes.
6. Translate the existing browser GUI style and tool behavior into Eve style
daemon-owned `.cc` witness using `cultcache-py`.
4. Publish Repixelizer daemon health to Idunn over CultNet/RUDP from the live
GUI process using `cultnet_py`, with a dedicated Repixelizer health contract.
5. Publish runtime-owned command-boundary and transport-profile state beside
the witness so deploy/restart authority is typed instead of implicit ops lore.
6. Update the Yggdrasil deploy lane to ship the required CultLib Python package
snapshot with the app artifact and install it before the Repixelizer package.
7. Publish the witness through CultMesh.
8. Add an Eve DSL provider over the witness and existing health/config routes.
First read-only HTTP projection is live at `/eve/surface`.
9. Translate the existing browser GUI style and tool behavior into Eve style
tokens, canvas primitives, and command descriptors.
7. Lower the current website from the Eve surface instead of treating it as the
10. Lower the current website from the Eve surface instead of treating it as the
product UI owner.
8. Register the surface with Odin.
9. Only after the witness is stable, decide whether queue ownership should move
11. Register the surface with Odin.
12. Only after the witness is stable, decide whether queue ownership should move
from in-process memory into the typed store.
The invariant: Heimdall owns shared auth; Repixelizer owns jobs and artifacts.

1
pyproject.toml
View file

@ -14,6 +14,7 @@ authors = [
]
dependencies = [
"fastapi>=0.115.0",
"msgpack>=1.0",
"numpy>=2.1.0",
"Pillow>=10.4.0",
"PyJWT[crypto]>=2.10.1",

19
src/repixelizer/cli.py
View file

@ -6,6 +6,7 @@ import sys
from .benchmark import run_roundtrip_benchmark
from .compare import run_compare
from .corpus import prepare_corpus
from .eve_surface import main as eve_surface_main
from .gui import main as gui_main
from .pipeline import run_pipeline
from .spritesheet import run_spritesheet
@ -142,6 +143,13 @@ def build_parser() -> argparse.ArgumentParser:
witness_parser.add_argument("--public-base-url", default=None, help="Hosted browser lowering base URL.")
witness_parser.add_argument("--cc-witness-path", default=None, help="Reserved CultCache .cc witness path.")
witness_parser.add_argument("--eve-surface-key", default=None, help="Planned CultMesh Eve surface key.")
eve_surface_parser = subparsers.add_parser(
"eve-surface",
help="Emit the read-only Repixelizer Eve surface projection.",
)
eve_surface_parser.add_argument("--out", default=None, help="Write JSON to this path instead of stdout.")
eve_surface_parser.add_argument("--updated-at", default=None, help="Override the surface timestamp.")
eve_surface_parser.add_argument("--public-base-url", default=None, help="Hosted browser lowering base URL.")
gui_parser = subparsers.add_parser("gui", help="Launch the web GUI.")
gui_parser.add_argument("--host", default="127.0.0.1", help="Host interface to bind")
gui_parser.add_argument("--port", type=int, default=8000, help="Port to bind")
@ -159,6 +167,7 @@ def main(argv: list[str] | None = None) -> int:
"tune",
"prepare-corpus",
"witness-advertisement",
"eve-surface",
"gui",
"-h",
"--help",
@ -254,6 +263,16 @@ def main(argv: list[str] | None = None) -> int:
if value is not None:
witness_args.extend([option, str(value)])
return witness_advertisement_main(witness_args)
if command == "eve-surface":
eve_surface_args: list[str] = []
for option, value in (
("--out", args.out),
("--updated-at", args.updated_at),
("--public-base-url", args.public_base_url),
):
if value is not None:
eve_surface_args.extend([option, str(value)])
return eve_surface_main(eve_surface_args)
if command == "gui":
return gui_main(host=args.host, port=args.port, reload=args.reload)

62
src/repixelizer/cultlib_support.py Normal file
View file

@ -0,0 +1,62 @@
from __future__ import annotations
import importlib
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any
@dataclass(frozen=True)
class CultLibBindings:
CultCache: Any
CultMesh: Any
SingleFileMessagePackBackingStore: Any
define_database_entry_type: Any
document_put_raw: Any
def _candidate_roots() -> list[Path]:
repo_root = Path(__file__).resolve().parents[2]
return [
repo_root.parent / "CultLib" / "packages" / "cultcache-py" / "src",
repo_root.parent / "cultcache-py" / "src",
]
def _ensure_cultlib_on_path() -> None:
for candidate in _candidate_roots():
if not candidate.exists():
continue
candidate_text = str(candidate)
if candidate_text not in sys.path:
sys.path.insert(0, candidate_text)
return
raise RuntimeError(
"Repixelizer verse runtime requires cultcache-py/cultnet_py. "
"Install E:\\Projects\\CultLib\\packages\\cultcache-py or add it to PYTHONPATH."
)
def _import_bindings() -> CultLibBindings:
cultcache_py = importlib.import_module("cultcache_py")
cultmesh_py = importlib.import_module("cultmesh_py")
cultnet_py = importlib.import_module("cultnet_py")
return CultLibBindings(
CultCache=cultcache_py.CultCache,
CultMesh=cultmesh_py.CultMesh,
SingleFileMessagePackBackingStore=cultcache_py.SingleFileMessagePackBackingStore,
define_database_entry_type=cultcache_py.define_database_entry_type,
document_put_raw=cultnet_py.document_put_raw,
)
def load_cultlib() -> CultLibBindings:
try:
return _import_bindings()
except ModuleNotFoundError:
_ensure_cultlib_on_path()
return _import_bindings()
__all__ = ["CultLibBindings", "load_cultlib"]

591
src/repixelizer/eve_surface.py Normal file
View file

@ -0,0 +1,591 @@
from __future__ import annotations
import argparse
import json
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, TextIO
from .access import AccessController
from .gui import HostedDemoConfig
EVE_SURFACE_SCHEMA = "gamecult.eve.surface.v1"
PROVIDER_ID = "repixelizer"
PROVIDER_KIND = "repixelizer.product"
DEFAULT_PUBLIC_BASE_URL = "https://repixelizer.gamecult.org"
def utc_now_iso() -> str:
return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def _text(element_id: str, text: str, *, role: str = "body", **props: Any) -> dict[str, Any]:
return {
"id": element_id,
"kind": "text",
"role": role,
"text": text,
"props": {"text": text, "role": role, **props},
"children": [],
}
def _card(element_id: str, title: str, children: list[dict[str, Any]], *, tone: str = "panel") -> dict[str, Any]:
return {
"id": element_id,
"kind": "card",
"text": title,
"role": "group",
"style": {"tone": tone, "variant": "repixelizer-retro-card"},
"props": {"title": title, "styleProfile": "repixelizer.retro.pixel"},
"children": children,
}
def _leaf(element_id: str, kind: str, *, text: str | None = None, role: str = "body", **props: Any) -> dict[str, Any]:
payload = {"text": text, "role": role, **props}
if text is None:
payload.pop("text")
return {
"id": element_id,
"kind": kind,
"role": role,
"text": text or props.get("label") or props.get("title") or element_id,
"props": payload,
"children": [],
}
def _button(element_id: str, label: str, *, command_id: str, **props: Any) -> dict[str, Any]:
return _leaf(element_id, "control.button", role="command", label=label, commandId=command_id, **props)
def _field(element_id: str, label: str, *, value: str, field_kind: str = "input.number", **props: Any) -> dict[str, Any]:
return _leaf(element_id, field_kind, role="field", label=label, value=value, **props)
def _pane(element_id: str, title: str, children: list[dict[str, Any]], *, direction: str = "vertical") -> dict[str, Any]:
return {
"id": element_id,
"kind": "pane",
"text": title,
"role": "section",
"layout": {"direction": direction, "padding": 10},
"style": {"tone": "warm", "variant": "repixelizer-retro-pane"},
"props": {"title": title, "styleProfile": "repixelizer.retro.pixel"},
"children": children,
}
def _command_card(
element_id: str,
title: str,
body: str,
*,
command_id: str,
route: str,
method: str,
) -> dict[str, Any]:
element = _card(
element_id,
title,
[
_text(f"{element_id}.body", body),
_text(f"{element_id}.route", f"{method} {route}", role="mono", route=route, method=method),
],
tone="cool",
)
element["commandId"] = command_id
element["props"]["commandId"] = command_id
element["props"]["transport"] = {"kind": "http", "route": route, "method": method}
return element
def _styles(public_base_url: str) -> dict[str, Any]:
base = public_base_url.rstrip("/")
return {
"profile": "repixelizer.retro.pixel",
"tokens": {
"fontTitle": "\"Press Start 2P\", \"VT323\", monospace",
"fontBody": "\"VT323\", \"Courier New\", monospace",
"fontMono": "\"VT323\", \"Courier New\", monospace",
"colorBackgroundTop": "#01040b",
"colorBackgroundMid": "#020813",
"colorBackgroundBottom": "#01050d",
"colorShellSurfaceTop": "#061424",
"colorShellSurfaceBottom": "#030b16",
"colorPanel": "#0a2239",
"colorPanelAlt": "#061627",
"colorTextBright": "#f7efd8",
"colorText": "#efe5c9",
"colorMuted": "#d3bb7f",
"colorAccent": "#ffd84a",
"colorAccentStrong": "#ffae1a",
"colorPanelBorder": "#0a2740",
"colorPanelBorderDeep": "#06192a",
"colorCornerAccent": "#ffb72a",
"colorShadow": "rgba(0, 0, 0, 0.55)",
"colorPixelGrid": "rgba(0, 0, 0, 0.3)",
"imageRendering": "pixelated",
"borderWidthPx": 4,
"borderBottomWidthPx": 5,
"cornerAccentPx": 12,
"scanlineOverlay": True,
"dashedInnerFrame": True,
"buttonGlyph": ">",
},
"assets": {
"fontCss": "https://fonts.googleapis.com/css2?family=Press+Start+2P&family=VT323&display=swap",
"landingInput": f"{base}/app/landing-assets/character-input.png",
"landingOutput": f"{base}/app/landing-assets/character-repixelized.png",
"appUrl": f"{base}/app/",
},
"loweringHints": {
"css": "Use the tokens as CSS custom properties and retain scanline/pixel-grid overlays.",
"android-native": "Load packaged or downloadable Press Start 2P and VT323 fonts, map style tokens before hard-coded tones, and render asset images instead of asset labels.",
"tui": "Preserve title/body font distinction as role metadata; canvas editing lowers to artifact refs and command intents only.",
},
}
def _runtime_cards(
*,
config: HostedDemoConfig,
auth_payload: dict[str, Any],
queue_summary: dict[str, Any],
) -> list[dict[str, Any]]:
provider_labels = []
for provider in auth_payload.get("providers", []):
if isinstance(provider, dict):
label = provider.get("label") or provider.get("slug")
if isinstance(label, str) and label:
provider_labels.append(label)
elif isinstance(provider, str):
provider_labels.append(provider)
queue_copy = (
f"{queue_summary.get('queueDepth', 0)} active or waiting / "
f"{queue_summary.get('queueCapacity', config.queue_capacity)} capacity"
)
return [
_card(
"repixelizer.runtime.config",
"Runtime",
[
_text("repixelizer.runtime.hosted", f"hostedDemo: {config.hosted_demo}", role="mono"),
_text("repixelizer.runtime.steps", f"defaultSteps: {config.default_steps} maxSteps: {config.max_steps}", role="mono"),
_text("repixelizer.runtime.dimensions", f"maxOutputDimension: {config.max_output_dimension}px", role="mono"),
],
),
_card(
"repixelizer.runtime.queue",
"Queue",
[
_text("repixelizer.queue.depth", queue_copy, role="strong"),
_text("repixelizer.queue.waiting", f"waitingCount: {queue_summary.get('waitingCount', 0)}", role="mono"),
_text("repixelizer.queue.active", f"hasActiveJob: {queue_summary.get('hasActiveJob', False)}", role="mono"),
],
),
_card(
"repixelizer.runtime.auth",
"Access",
[
_text("repixelizer.auth.mode", f"mode: {auth_payload.get('mode', 'off')}", role="mono"),
_text("repixelizer.auth.required", f"required: {bool(auth_payload.get('required'))}", role="mono"),
_text("repixelizer.auth.providers", f"providers: {', '.join(provider_labels) or 'none'}"),
],
),
]
def _landing_gallery(base: str) -> dict[str, Any]:
return _pane(
"repixelizer.product.landing",
"Hosted demo",
[
_text(
"repixelizer.product.pitch",
"You need pixel art for a project, but pixel art is a craft and your favorite AI is mostly vibes in a trench coat.",
role="strong",
),
{
"id": "repixelizer.product.gallery",
"kind": "partition",
"role": "gallery",
"layout": {"direction": "horizontal", "gap": 12},
"props": {"split": "x", "gap": 12, "role": "product.gallery"},
"children": [
_card(
"repixelizer.product.input",
"The AI gives you this",
[
_leaf(
"repixelizer.product.input.image",
"image.preview",
label="AI-generated hazmat character before repixelizing",
src=f"{base}/app/landing-assets/character-input.png",
imageRendering="pixelated",
aspectRatio="1/1",
),
_text("repixelizer.product.input.copy", "Looks charming from orbit. Suspiciously charming."),
],
tone="cool",
),
_card(
"repixelizer.product.closeup",
"Then you zoom in",
[
_leaf(
"repixelizer.product.closeup.image",
"image.preview",
label="Zoomed close-up showing fake pixel-art artifacts",
src=f"{base}/app/landing-assets/character-input.png",
crop="42% 63%",
zoom=3.45,
imageRendering="pixelated",
aspectRatio="1/1",
),
_text("repixelizer.product.closeup.copy", "Smears, half-cells, anti-grid nonsense. The crime scene has opinions."),
],
tone="cool",
),
_card(
"repixelizer.product.output",
"Repixelizer spits out this",
[
_leaf(
"repixelizer.product.output.image",
"image.preview",
label="Repixelized hazmat character output",
src=f"{base}/app/landing-assets/character-repixelized.png",
imageRendering="pixelated",
aspectRatio="1/1",
),
_text("repixelizer.product.output.copy", "Real cells on a real lattice, ready for cleanup or export."),
],
tone="cool",
),
],
},
],
)
def _workspace_surface(config: HostedDemoConfig, queue_summary: dict[str, Any]) -> dict[str, Any]:
return {
"id": "repixelizer.workspace",
"kind": "partition",
"role": "workspace",
"layout": {"direction": "horizontal", "gap": 12},
"props": {"split": "x", "gap": 12, "role": "repixelizer.workspace"},
"children": [
_pane(
"repixelizer.workspace.viewer",
"Pipeline Stages",
[
_leaf(
"repixelizer.workspace.status",
"status.stage",
label="Idle",
stage="Waiting for input",
detail="Choose a file, then run the machine.",
),
{
"id": "repixelizer.workspace.compare",
"kind": "partition",
"role": "comparison",
"layout": {"direction": "horizontal", "gap": 10},
"props": {"split": "x", "gap": 10, "role": "comparison"},
"children": [
_leaf("repixelizer.workspace.input.canvas", "canvas.preview", label="Input", state="empty"),
_leaf("repixelizer.workspace.output.canvas", "canvas.preview", label="Output", state="empty"),
],
},
_card(
"repixelizer.workspace.inspect",
"Zoom",
[
_text("repixelizer.workspace.inspect.copy", "Hold either preview to inspect both panels together."),
_field("repixelizer.workspace.inspect.zoom", "Inspect Zoom", value="8x", field_kind="input.select"),
],
),
],
),
_pane(
"repixelizer.workspace.editor",
"Pixel Fixes",
[
{
"id": "repixelizer.workspace.editor.actions",
"kind": "partition",
"role": "toolbar",
"layout": {"direction": "horizontal", "gap": 8},
"props": {"split": "x", "gap": 8, "role": "toolbar"},
"children": [
_button("repixelizer.workspace.reset", "Reset Output", command_id="repixelizer.editor.reset"),
_button("repixelizer.workspace.export", "Export PNG", command_id="repixelizer.editor.export"),
],
},
{
"id": "repixelizer.workspace.paint",
"kind": "partition",
"role": "paint-controls",
"layout": {"direction": "horizontal", "gap": 6},
"props": {"split": "x", "gap": 6, "role": "paint-controls"},
"children": [
_leaf("repixelizer.workspace.swatch", "color.swatch", label="Paint", value="#ffffffff"),
_field("repixelizer.workspace.paint.r", "R", value="255"),
_field("repixelizer.workspace.paint.g", "G", value="255"),
_field("repixelizer.workspace.paint.b", "B", value="255"),
_field("repixelizer.workspace.paint.a", "A", value="255"),
_field("repixelizer.workspace.zoom", "Zoom", value="12x", field_kind="control.range", min=4, max=32),
_leaf("repixelizer.workspace.grid", "control.toggle", label="Grid", value=True),
],
},
_leaf("repixelizer.workspace.editor.canvas", "canvas.editor", label="Editable output canvas", state="empty"),
],
),
_pane(
"repixelizer.workspace.sidebar",
"Run Controls",
[
_card(
"repixelizer.workspace.input",
"Input",
[
_text("repixelizer.workspace.input.copy", "Start with the ugly input."),
_leaf("repixelizer.workspace.file", "input.file", label="Choose File", accept="image/png,image/*"),
_leaf("repixelizer.workspace.dropzone", "dropzone", label="PNG, sprite scrap, cursed logo. Whatever started this."),
],
),
_card(
"repixelizer.workspace.controls",
"Controls",
[
_field("repixelizer.workspace.target.size", "Target Size", value="auto"),
_field("repixelizer.workspace.target.width", "Target Width", value="auto"),
_field("repixelizer.workspace.target.height", "Target Height", value="auto"),
_field("repixelizer.workspace.steps", "Steps", value=str(config.default_steps)),
_button("repixelizer.workspace.run", "Run The Machine", command_id="repixelizer.job.submit", route="/api/jobs", method="POST"),
],
),
_card(
"repixelizer.workspace.queue",
"Queue",
[
_leaf("repixelizer.workspace.queue.depth", "metric", label="Queue", value=queue_summary.get("queueDepth", 0), max=queue_summary.get("queueCapacity", config.queue_capacity)),
_leaf("repixelizer.workspace.queue.worker", "metric", label="Worker", value=1 if queue_summary.get("hasActiveJob") else 0, max=1),
],
),
_card(
"repixelizer.workspace.lattice",
"Lattice",
[_text("repixelizer.workspace.lattice.copy", "No lattice picked yet.")],
),
],
),
],
}
def _lowering_comparison() -> dict[str, Any]:
return {
"periwinkleAndroidKotlin": {
"source": "E:\\Projects\\Eve\\android\\app\\src\\main\\java\\org\\gamecult\\eve\\MainActivity.kt",
"status": "surface-tree-renderer-present-style-token-gap",
"keeps": [
"surface root traversal",
"pane/card/text hierarchy",
"TextView body content",
"basic ProgressBar metric rendering",
"node-bound command taps",
],
"losesUntilRendererParity": [
"Press Start 2P and VT323 font selection",
"Repixelizer shell frame, scanlines, pixel grid, corner accents, and chunky border widths",
"landing/app image assets; current Android lowerer prints asset refs when images are not explicitly implemented",
"upload picker, comparison canvas, pan/zoom inspection, eyedropper, and paint controls",
"provider style tokens; current Kotlin toneColor/textColorFor/textSizeFor are hard-coded",
],
"minimumParityCuts": [
"map surface.styles.tokens before toneColor/textColorFor/textSizeFor",
"add packaged pixel fonts or font download cache",
"render image assetUri/assetRef as ImageView with pixelated sampling",
"lower command controls to Android upload/document picker and job route calls",
],
}
}
def build_repixelizer_eve_surface(
*,
updated_at: str | None = None,
public_base_url: str = DEFAULT_PUBLIC_BASE_URL,
config: HostedDemoConfig | None = None,
access_controller: AccessController | None = None,
queue_summary: dict[str, Any] | None = None,
) -> dict[str, Any]:
config = config or HostedDemoConfig.from_env()
access_controller = access_controller or AccessController.from_env(hosted_demo=config.hosted_demo)
auth_payload = access_controller.public_payload()
queue_summary = queue_summary or {
"queueDepth": 0,
"waitingCount": 0,
"queueCapacity": config.queue_capacity,
"hasActiveJob": False,
"activeStatus": None,
}
updated_at = updated_at or utc_now_iso()
base = public_base_url.rstrip("/")
root = {
"id": "repixelizer.surface.root",
"kind": "surface",
"text": "Repixelizer",
"role": "root",
"layout": {"direction": "vertical", "padding": 14},
"style": {"tone": "warm", "variant": "repixelizer-retro-shell"},
"props": {
"title": "Repixelizer",
"subtitle": "Fake pixel art goes in. Real grid-aligned sprite art comes out.",
"styleProfile": "repixelizer.retro.pixel",
"browserLowering": f"{base}/app/",
"frontPage": f"{base}/",
},
"children": [
_pane(
"repixelizer.hero",
"Repixelizer hosted demo",
[
_text(
"repixelizer.hero.copy",
"Force fake pixel art back onto a real grid, then inspect each solver step without sanding off the crunchy old-machine charm.",
),
_command_card(
"repixelizer.hero.open",
"Open demo",
"Browser lowering remains the full interactive app until native Eve gains upload and canvas editing parity.",
command_id="repixelizer.web.open",
route="/app/",
method="GET",
),
],
),
_landing_gallery(base),
_workspace_surface(config, queue_summary),
_pane(
"repixelizer.workflow",
"Single page app",
[
_command_card(
"repixelizer.workflow.submit",
"Upload image",
"Submit a PNG/JPEG/WebP/GIF into the GUI job queue; queue ownership stays in repixelizer.gui.",
command_id="repixelizer.job.submit",
route="/api/jobs",
method="POST",
),
_card(
"repixelizer.workflow.inspect",
"Inspect solver",
[
_text("repixelizer.inspect.source", "source, edge map, lattice overlay, phase field, cleaned output"),
_text("repixelizer.inspect.canvas", "comparison canvas and editor lower to native only after Periwinkle gets image/control parity"),
],
),
_command_card(
"repixelizer.workflow.cancel",
"Cancel own job",
"The command intent delegates to the existing owner-bound DELETE route.",
command_id="repixelizer.job.cancel_own",
route="/api/jobs/{job_id}",
method="DELETE",
),
],
),
_pane("repixelizer.runtime", "Live projection", _runtime_cards(config=config, auth_payload=auth_payload, queue_summary=queue_summary)),
],
}
return {
"type": "surface-state",
"schema": EVE_SURFACE_SCHEMA,
"providerId": PROVIDER_ID,
"providerKind": PROVIDER_KIND,
"title": "Repixelizer",
"version": 1,
"updatedAt": updated_at,
"surface": {
"id": "repixelizer.operator.surface",
"schema": EVE_SURFACE_SCHEMA,
"title": "Repixelizer",
"root": root,
"styles": _styles(public_base_url),
},
"commands": [
{
"command": "repixelizer.web.open",
"surfaceId": "repixelizer.operator.surface",
"transport": "http",
"route": "/app/",
"method": "GET",
"authority": "browser-lowering",
},
{
"command": "repixelizer.job.submit",
"surfaceId": "repixelizer.operator.surface",
"transport": "http",
"route": "/api/jobs",
"method": "POST",
"authority": "repixelizer-route-policy",
},
{
"command": "repixelizer.job.cancel_own",
"surfaceId": "repixelizer.operator.surface",
"transport": "http",
"route": "/api/jobs/{job_id}",
"method": "DELETE",
"authority": "repixelizer-job-owner",
},
],
"runtime": {
"hostedDemo": config.hosted_demo,
"queue": queue_summary,
"auth": auth_payload,
"limits": config.public_payload()["limits"],
},
"loweringComparison": _lowering_comparison(),
}
def write_repixelizer_eve_surface(
output: Path | None,
*,
updated_at: str | None = None,
public_base_url: str = DEFAULT_PUBLIC_BASE_URL,
stream: TextIO = sys.stdout,
) -> None:
payload = build_repixelizer_eve_surface(updated_at=updated_at, public_base_url=public_base_url)
encoded = json.dumps(payload, indent=2, sort_keys=True)
if output is None:
stream.write(f"{encoded}\n")
return
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(f"{encoded}\n", encoding="utf-8")
def build_eve_surface_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="repixelize eve-surface",
description="Emit Repixelizer's read-only Eve surface projection.",
)
parser.add_argument("--out", type=Path, default=None, help="Write JSON to this path instead of stdout.")
parser.add_argument("--updated-at", default=None, help="Override the surface timestamp.")
parser.add_argument("--public-base-url", default=DEFAULT_PUBLIC_BASE_URL, help="Hosted browser lowering base URL.")
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_eve_surface_parser()
args = parser.parse_args(argv)
write_repixelizer_eve_surface(args.out, updated_at=args.updated_at, public_base_url=args.public_base_url)
return 0

78
src/repixelizer/gui.py
View file

@ -24,6 +24,7 @@ from .inference import inference_to_json
from .observe import PipelineCancelled
from .pipeline import _resolve_requested_target_dims, run_pipeline_rgba
from .types import InferenceResult, PaletteResult
from .verse_state import RepixelizerVerseRuntime, RepixelizerVerseRuntimeConfig
LOGGER = logging.getLogger(__name__)
@ -788,6 +789,57 @@ class GuiJobManager:
"activeStatus": None if self._active_job_id is None else self.jobs[self._active_job_id].status,
}
def snapshot_runtime_state(self, *, max_events_per_job: int = 32) -> dict[str, Any]:
with self._condition:
queue_depth = self._queue_depth_locked()
waiting_count = len(self._queued_job_ids)
jobs = []
for job in sorted(self.jobs.values(), key=lambda item: item.created_at):
jobs.append(
{
"jobId": job.job_id,
"filename": job.filename,
"status": job.status,
"error": job.error,
"createdAt": job.created_at,
"runningAt": job.running_at,
"lastHeartbeatAt": job.last_heartbeat_at,
"eventCount": len(job.events),
"accountId": job.account_id,
"sessionId": job.session_id,
"accessRevision": job.access_revision,
"displayName": job.display_name,
"authMode": job.auth_mode,
"identityProvider": job.identity_provider,
"queuePosition": self._queue_position_locked(job.job_id),
"queueDepth": queue_depth,
"waitingCount": waiting_count,
"spoolPath": str(job.spool_path),
"options": dict(job.options),
"events": [
{
"id": int(event["id"]),
"event": str(event["event"]),
"timestamp": float(event["timestamp"]),
"payload": dict(event["payload"]),
}
for event in job.events[-max_events_per_job:]
],
}
)
return {
"queue": {
"queueDepth": queue_depth,
"waitingCount": waiting_count,
"queueCapacity": self.config.queue_capacity,
"hasActiveJob": self._active_job_id is not None,
"activeStatus": None if self._active_job_id is None else self.jobs[self._active_job_id].status,
"activeJobId": self._active_job_id,
"queuedJobIds": list(self._queued_job_ids),
},
"jobs": jobs,
}
def heartbeat(self, job_id: str) -> dict[str, Any] | None:
with self._condition:
job = self.jobs.get(job_id)
@ -1044,13 +1096,21 @@ def create_app():
config = HostedDemoConfig.from_env()
manager = GuiJobManager(config)
access_controller = AccessController.from_env(hosted_demo=config.hosted_demo)
verse_runtime = RepixelizerVerseRuntime(
RepixelizerVerseRuntimeConfig.from_env(config),
config,
access_controller,
manager,
)
@contextlib.asynccontextmanager
async def lifespan(_app):
manager.start()
try:
verse_runtime.start()
yield
finally:
verse_runtime.stop()
manager.stop()
app = FastAPI(title="Repixelizer GUI", version="0.1.0", lifespan=lifespan)
@ -1117,8 +1177,8 @@ def create_app():
return await call_with_subject(subject)
@app.get("/api/health")
def health() -> dict[str, str]:
return {"status": "ok"}
def health():
return JSONResponse(verse_runtime.build_health_payload())
@app.get("/api/config")
def get_config():
@ -1191,6 +1251,20 @@ def create_app():
def get_queue():
return JSONResponse(manager.get_queue_summary())
@app.get("/eve/surface")
def get_eve_surface(request: Request):
from .eve_surface import build_repixelizer_eve_surface
base_url = str(request.base_url).rstrip("/")
return JSONResponse(
build_repixelizer_eve_surface(
public_base_url=base_url,
config=config,
access_controller=access_controller,
queue_summary=manager.get_queue_summary(),
)
)
@app.get("/api/jobs/{job_id}")
def get_job(job_id: str):
job = manager.get_job(job_id)

739
src/repixelizer/verse_state.py Normal file
View file

@ -0,0 +1,739 @@
from __future__ import annotations
import contextlib
import logging
import os
import time
from dataclasses import dataclass
from datetime import UTC, datetime
from functools import lru_cache
from pathlib import Path
from threading import Event, Lock, Thread
from typing import TYPE_CHECKING, Any
from .access import AccessController
from .cultlib_support import CultLibBindings, load_cultlib
if TYPE_CHECKING:
from .gui import HostedDemoConfig
LOGGER = logging.getLogger(__name__)
CULTNET_RUDP_PROTOCOL_ID = "cultnet.transport.rudp.v0"
IDUNN_HEALTH_RUDP_CONNECTION_ID = 0x1D0D_0001
IDUNN_HEALTH_CHANNEL_ID = "idunn.daemon_health"
DEFAULT_IDUNN_HEALTH_CONTRACT = "repixelizer.cultnet-rudp-service-health"
DEFAULT_CULTCACHE_FILENAME = "repixelizer.service.cc"
PUBLISH_INTERVAL_SECONDS = 5.0
MAX_EVENT_RECORDS_PER_JOB = 32
MAX_EVENT_LIST_ITEMS = 12
MAX_EVENT_DICT_ITEMS = 12
IMAGEISH_KEYS = {
"cleanedImage",
"diagnostics",
"displacementImage",
"edgeMapImage",
"heatmapImage",
"inference",
"latticeImage",
"lossHistory",
"outputImage",
"palette",
"runSummary",
"samplingOverlayImage",
"signalImage",
"sourceImage",
}
def _repo_root() -> Path:
return Path(__file__).resolve().parents[2]
def _utc_now_iso() -> str:
return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
@lru_cache(maxsize=1)
def _cultlib() -> CultLibBindings:
return load_cultlib()
def _define_database_entry_type(*args: Any, **kwargs: Any) -> Any:
return _cultlib().define_database_entry_type(*args, **kwargs)
def _normalize_public_base_url(value: str | None) -> str:
if value and value.strip():
return value.rstrip("/")
return "https://repixelizer.gamecult.org"
def _default_cultcache_path(hosted_config: HostedDemoConfig) -> Path:
return hosted_config.spool_dir.parent / "cultcache" / DEFAULT_CULTCACHE_FILENAME
def _health_document():
return _define_database_entry_type(
"idunn.daemon_health",
[
("daemon_id", 0),
("state", 1),
("detail", 2),
("observed_at", 3),
("health_contract", 4),
("publication_source", 5, ""),
("transport", 6, ""),
],
schema_id="idunn.daemon_health",
schema_name="idunn.daemon_health",
schema_version="idunn.daemon_health.v1",
)
def _provider_advertisement_document():
return _define_database_entry_type(
"gamecult.eve.provider_advertisement",
[("value", 0)],
schema_id="gamecult.eve.provider_advertisement.v1",
schema_name="gamecult.eve.provider_advertisement",
schema_version="gamecult.eve.provider_advertisement.v1",
)
def _eve_surface_document():
return _define_database_entry_type(
"gamecult.eve.surface_state",
[
("provider_id", 0),
("title", 1),
("version", 2),
("updated_at", 3),
("surface", 4),
],
schema_id="gamecult.eve.surface_state.v1",
schema_name="gamecult.eve.surface_state",
schema_version="gamecult.eve.surface_state.v1",
)
def _runtime_config_document():
return _define_database_entry_type(
"repixelizer.runtime_config",
[
("daemon_id", 0),
("hosted_demo", 1),
("queue_capacity", 2),
("show_queue_panel", 3),
("max_upload_bytes", 4),
("max_input_dimension", 5),
("max_output_dimension", 6),
("default_steps", 7),
("max_steps", 8),
("heartbeat_interval_seconds", 9),
("stale_after_seconds", 10),
("spool_path", 11),
("public_base_url", 12),
],
schema_id="repixelizer.runtime_config",
schema_name="repixelizer.runtime_config",
schema_version="repixelizer.runtime_config.v0",
)
def _auth_projection_document():
return _define_database_entry_type(
"repixelizer.auth_projection",
[
("daemon_id", 0),
("enabled", 1),
("mode", 2),
("required", 3),
("protect_queue", 4),
("providers", 5),
("updated_at", 6),
],
schema_id="repixelizer.auth_projection",
schema_name="repixelizer.auth_projection",
schema_version="repixelizer.auth_projection.v0",
)
def _queue_snapshot_document():
return _define_database_entry_type(
"repixelizer.queue_snapshot",
[
("daemon_id", 0),
("queue_depth", 1),
("waiting_count", 2),
("queue_capacity", 3),
("has_active_job", 4),
("active_status", 5, None),
("active_job_id", 6, None),
("queued_job_ids", 7),
("updated_at", 8),
],
schema_id="repixelizer.queue_snapshot",
schema_name="repixelizer.queue_snapshot",
schema_version="repixelizer.queue_snapshot.v0",
)
def _job_document():
return _define_database_entry_type(
"repixelizer.job",
[
("job_id", 0),
("status", 1),
("filename", 2),
("created_at", 3),
("running_at", 4, None),
("last_heartbeat_at", 5),
("queue_position", 6, None),
("queue_depth", 7),
("waiting_count", 8),
("event_count", 9),
("error", 10, None),
("account_id", 11, None),
("session_id", 12, None),
("access_revision", 13, None),
("display_name", 14, None),
("auth_mode", 15, None),
("identity_provider", 16, None),
("target_size", 17, None),
("target_width", 18, None),
("target_height", 19, None),
("steps", 20, None),
("spool_path", 21),
("updated_at", 22),
],
schema_id="repixelizer.job",
schema_name="repixelizer.job",
schema_version="repixelizer.job.v0",
)
def _job_event_document():
return _define_database_entry_type(
"repixelizer.job_event",
[
("job_id", 0),
("event_id", 1),
("event_name", 2),
("timestamp", 3),
("payload_summary", 4),
("updated_at", 5),
],
schema_id="repixelizer.job_event",
schema_name="repixelizer.job_event",
schema_version="repixelizer.job_event.v0",
)
def _command_boundary_document():
return _define_database_entry_type(
"repixelizer.command_boundary",
[
("daemon_id", 0),
("boundary_id", 1),
("updated_at", 2),
("owner", 3),
("deploy_authority", 4),
("health_authority", 5),
("queue_authority", 6),
("compatibility_witnesses", 7),
("forbidden_writers", 8),
],
schema_id="repixelizer.command_boundary.v1",
schema_name="repixelizer.command_boundary",
schema_version="repixelizer.command_boundary.v1",
)
def _transport_profile_document():
return _define_database_entry_type(
"repixelizer.transport_profile",
[
("daemon_id", 0),
("profile_id", 1),
("updated_at", 2),
("target_transport", 3),
("current_transport", 4),
("health_transport", 5),
("state_transport", 6),
("renderer_transport", 7),
("health_contract", 8),
("cut_line", 9),
],
schema_id="repixelizer.transport_profile.v1",
schema_name="repixelizer.transport_profile",
schema_version="repixelizer.transport_profile.v1",
)
@dataclass(frozen=True)
class RepixelizerVerseRuntimeConfig:
enabled: bool
daemon_id: str
cultcache_path: Path
idunn_rudp_health: str | None
idunn_health_contract: str
public_base_url: str
workspace_root: Path
@classmethod
def from_env(cls, hosted_config: HostedDemoConfig) -> "RepixelizerVerseRuntimeConfig":
hosted_default_daemon = "yggdrasil-repixelizer" if hosted_config.hosted_demo else "repixelizer"
daemon_id = os.getenv("GC_ACCESS_IDUNN_DAEMON", hosted_default_daemon).strip() or hosted_default_daemon
cultcache_path = Path(os.getenv("GC_ACCESS_CULTCACHE_PATH", str(_default_cultcache_path(hosted_config)))).expanduser()
idunn_rudp_health = os.getenv("GC_ACCESS_IDUNN_RUDP_HEALTH")
if idunn_rudp_health is not None:
idunn_rudp_health = idunn_rudp_health.strip() or None
idunn_health_contract = os.getenv("GC_ACCESS_IDUNN_HEALTH_CONTRACT", DEFAULT_IDUNN_HEALTH_CONTRACT).strip()
enabled = (
os.getenv("REPIXELIZER_VERSE_RUNTIME", "").strip().lower() in {"1", "true", "yes", "on"}
or "GC_ACCESS_CULTCACHE_PATH" in os.environ
or "GC_ACCESS_IDUNN_RUDP_HEALTH" in os.environ
or "GC_ACCESS_IDUNN_DAEMON" in os.environ
)
return cls(
enabled=enabled,
daemon_id=daemon_id,
cultcache_path=cultcache_path,
idunn_rudp_health=idunn_rudp_health,
idunn_health_contract=idunn_health_contract,
public_base_url=_normalize_public_base_url(os.getenv("GC_ACCESS_APP_PUBLIC_BASE_URL")),
workspace_root=_repo_root(),
)
@dataclass(frozen=True)
class RepixelizerRuntimePulse:
updated_at: str
queue: dict[str, Any]
jobs: list[dict[str, Any]]
auth_payload: dict[str, Any]
class RepixelizerVerseRuntime:
def __init__(
self,
runtime_config: RepixelizerVerseRuntimeConfig,
hosted_config: HostedDemoConfig,
access_controller: AccessController,
manager: Any,
) -> None:
self.runtime_config = runtime_config
self.hosted_config = hosted_config
self.access_controller = access_controller
self.manager = manager
self._stop_event = Event()
self._thread: Thread | None = None
self._publish_lock = Lock()
self._documents = self._build_documents() if runtime_config.enabled else ()
self._document_by_type = {document.type: document for document in self._documents}
@property
def enabled(self) -> bool:
return self.runtime_config.enabled
def start(self) -> None:
if not self.enabled or self._thread is not None:
return
self.publish_once()
self._stop_event.clear()
self._thread = Thread(target=self._loop, name="repixelizer-verse-runtime", daemon=True)
self._thread.start()
def stop(self) -> None:
if self._thread is None:
return
self._stop_event.set()
self._thread.join(timeout=2.0)
self._thread = None
def build_health_payload(self) -> dict[str, Any]:
queue_summary = self.manager.get_queue_summary()
auth_payload = self.access_controller.public_payload()
return {
"status": "ok",
"verseRuntime": "enabled" if self.enabled else "disabled",
"cultCachePath": str(self.runtime_config.cultcache_path) if self.enabled else None,
"idunnDaemon": self.runtime_config.daemon_id,
"idunnHealthContract": self.runtime_config.idunn_health_contract,
"idunnRudpHealth": self.runtime_config.idunn_rudp_health,
"queue": queue_summary,
"auth": {
"enabled": bool(auth_payload.get("enabled")),
"mode": auth_payload.get("mode"),
"required": bool(auth_payload.get("required")),
},
}
def publish_once(self) -> None:
if not self.enabled:
return
with self._publish_lock:
pulse = self._build_pulse()
final_path = self.runtime_config.cultcache_path
temp_path = final_path.with_suffix(final_path.suffix + f".tmp-{os.getpid()}-{int(time.time() * 1000)}")
try:
cache = self._build_cache(temp_path)
records = self._build_records(pulse)
health_envelope = None
for document, key, value in records:
cache.put(document, key, value)
if document.type == "idunn.daemon_health":
health_envelope = cache.get_required_envelope(document, key)
final_path.parent.mkdir(parents=True, exist_ok=True)
temp_path.replace(final_path)
if health_envelope is not None and self.runtime_config.idunn_rudp_health:
self._publish_idunn_health(health_envelope.key, health_envelope.stored_at, health_envelope.payload)
finally:
with contextlib.suppress(FileNotFoundError):
temp_path.unlink()
def _loop(self) -> None:
while not self._stop_event.wait(PUBLISH_INTERVAL_SECONDS):
try:
self.publish_once()
except Exception:
LOGGER.exception("Repixelizer verse runtime publish failed.")
def _build_documents(self) -> tuple[Any, ...]:
return (
_provider_advertisement_document(),
_eve_surface_document(),
_runtime_config_document(),
_auth_projection_document(),
_queue_snapshot_document(),
_job_document(),
_job_event_document(),
_command_boundary_document(),
_transport_profile_document(),
_health_document(),
)
def _build_cache(self, store_path: Path) -> Any:
bindings = _cultlib()
builder = bindings.CultCache.builder()
for document in self._documents:
builder.register_document_type(document)
builder.add_generic_store(bindings.SingleFileMessagePackBackingStore(store_path))
cache = builder.build()
cache.pull_all_backing_stores()
return cache
def _document(self, document_type: str) -> Any:
return self._document_by_type[document_type]
def _build_pulse(self) -> RepixelizerRuntimePulse:
snapshot = self.manager.snapshot_runtime_state(max_events_per_job=MAX_EVENT_RECORDS_PER_JOB)
return RepixelizerRuntimePulse(
updated_at=_utc_now_iso(),
queue=snapshot["queue"],
jobs=snapshot["jobs"],
auth_payload=self.access_controller.public_payload(),
)
def _build_records(self, pulse: RepixelizerRuntimePulse) -> list[tuple[Any, str, Any]]:
from .eve_surface import build_repixelizer_eve_surface
queue = pulse.queue
provider_advertisement = self._build_provider_advertisement(pulse)
eve_surface = build_repixelizer_eve_surface(
updated_at=pulse.updated_at,
public_base_url=self.runtime_config.public_base_url,
config=self.hosted_config,
access_controller=self.access_controller,
queue_summary={
"queueDepth": queue["queueDepth"],
"waitingCount": queue["waitingCount"],
"queueCapacity": queue["queueCapacity"],
"hasActiveJob": queue["hasActiveJob"],
"activeStatus": queue["activeStatus"],
},
)
records: list[tuple[Any, str, Any]] = [
(self._document("gamecult.eve.provider_advertisement"), "repixelizer", {"value": provider_advertisement}),
(
self._document("gamecult.eve.surface_state"),
"repixelizer",
{
"provider_id": "repixelizer",
"title": "Repixelizer",
"version": int(eve_surface.get("version", 1)),
"updated_at": pulse.updated_at,
"surface": eve_surface,
},
),
(
self._document("repixelizer.runtime_config"),
self.runtime_config.daemon_id,
{
"daemon_id": self.runtime_config.daemon_id,
"hosted_demo": self.hosted_config.hosted_demo,
"queue_capacity": self.hosted_config.queue_capacity,
"show_queue_panel": self.hosted_config.show_queue_panel,
"max_upload_bytes": self.hosted_config.max_upload_bytes,
"max_input_dimension": self.hosted_config.max_input_dimension,
"max_output_dimension": self.hosted_config.max_output_dimension,
"default_steps": self.hosted_config.default_steps,
"max_steps": self.hosted_config.max_steps,
"heartbeat_interval_seconds": self.hosted_config.heartbeat_interval_seconds,
"stale_after_seconds": self.hosted_config.stale_after_seconds,
"spool_path": str(self.hosted_config.spool_dir),
"public_base_url": self.runtime_config.public_base_url,
},
),
(
self._document("repixelizer.auth_projection"),
self.runtime_config.daemon_id,
{
"daemon_id": self.runtime_config.daemon_id,
"enabled": bool(pulse.auth_payload.get("enabled")),
"mode": pulse.auth_payload.get("mode"),
"required": bool(pulse.auth_payload.get("required")),
"protect_queue": bool(pulse.auth_payload.get("protectQueue")),
"providers": pulse.auth_payload.get("providers", []),
"updated_at": pulse.updated_at,
},
),
(
self._document("repixelizer.queue_snapshot"),
self.runtime_config.daemon_id,
{
"daemon_id": self.runtime_config.daemon_id,
"queue_depth": queue["queueDepth"],
"waiting_count": queue["waitingCount"],
"queue_capacity": queue["queueCapacity"],
"has_active_job": queue["hasActiveJob"],
"active_status": queue["activeStatus"],
"active_job_id": queue["activeJobId"],
"queued_job_ids": queue["queuedJobIds"],
"updated_at": pulse.updated_at,
},
),
(
self._document("repixelizer.command_boundary"),
"repixelizer",
{
"daemon_id": self.runtime_config.daemon_id,
"boundary_id": "repixelizer",
"updated_at": pulse.updated_at,
"owner": "Repixelizer GUI runtime",
"deploy_authority": "idunn.yggdrasil-source-app.deploy",
"health_authority": "repixelizer.gui->idunn.daemon_health",
"queue_authority": "repixelizer.gui.GuiJobManager",
"compatibility_witnesses": [
"repixelizer-gui.service",
"GET /api/health",
"GET /api/config",
"nginx /app/ and /api/health host routing",
],
"forbidden_writers": [
"Idunn and Odin may observe Repixelizer daemon truth but do not mutate queue ownership.",
"The browser app, nginx, and systemd do not own queue or daemon health truth.",
],
},
),
(
self._document("repixelizer.transport_profile"),
"repixelizer",
{
"daemon_id": self.runtime_config.daemon_id,
"profile_id": "repixelizer",
"updated_at": pulse.updated_at,
"target_transport": CULTNET_RUDP_PROTOCOL_ID,
"current_transport": self._current_transport(),
"health_transport": CULTNET_RUDP_PROTOCOL_ID if self.runtime_config.idunn_rudp_health else "cultcache-store",
"state_transport": "cultcache.store.v1",
"renderer_transport": "browser-http-lowering",
"health_contract": self.runtime_config.idunn_health_contract,
"cut_line": self._transport_cut_line(),
},
),
(
self._document("idunn.daemon_health"),
self.runtime_config.daemon_id,
self._build_daemon_health_record(pulse),
),
]
for job in pulse.jobs:
records.append(
(
self._document("repixelizer.job"),
job["jobId"],
{
"job_id": job["jobId"],
"status": job["status"],
"filename": job["filename"],
"created_at": job["createdAt"],
"running_at": job["runningAt"],
"last_heartbeat_at": job["lastHeartbeatAt"],
"queue_position": job["queuePosition"],
"queue_depth": job["queueDepth"],
"waiting_count": job["waitingCount"],
"event_count": job["eventCount"],
"error": job["error"],
"account_id": job["accountId"],
"session_id": job["sessionId"],
"access_revision": job["accessRevision"],
"display_name": job["displayName"],
"auth_mode": job["authMode"],
"identity_provider": job["identityProvider"],
"target_size": job["options"].get("target_size"),
"target_width": job["options"].get("target_width"),
"target_height": job["options"].get("target_height"),
"steps": job["options"].get("steps"),
"spool_path": job["spoolPath"],
"updated_at": pulse.updated_at,
},
)
)
for event in job["events"]:
records.append(
(
self._document("repixelizer.job_event"),
f"{job['jobId']}:{event['id']}",
{
"job_id": job["jobId"],
"event_id": event["id"],
"event_name": event["event"],
"timestamp": event["timestamp"],
"payload_summary": _summarize_event_payload(event["payload"]),
"updated_at": pulse.updated_at,
},
)
)
return records
def _build_provider_advertisement(self, pulse: RepixelizerRuntimePulse) -> dict[str, Any]:
from .witness import build_provider_advertisement
witness_path = _display_witness_path(self.runtime_config.workspace_root, self.runtime_config.cultcache_path)
advertisement = build_provider_advertisement(
updated_at=pulse.updated_at,
public_base_url=self.runtime_config.public_base_url,
cc_witness_path=witness_path,
config=self.hosted_config,
access_controller=self.access_controller,
)
witness = dict(advertisement["witnesses"][0])
witness["kind"] = "cc-export-path-live"
witness["freshness"] = {"state": "fresh", "updatedAt": pulse.updated_at}
advertisement["witnesses"] = [witness]
advertisement["status"] = "daemon_live"
advertisement["runtime"] = {
**advertisement["runtime"],
"cultCachePath": str(self.runtime_config.cultcache_path),
"idunnDaemon": self.runtime_config.daemon_id,
"idunnHealthContract": self.runtime_config.idunn_health_contract,
"idunnRudpHealth": self.runtime_config.idunn_rudp_health,
"healthTransport": CULTNET_RUDP_PROTOCOL_ID if self.runtime_config.idunn_rudp_health else "cultcache-store",
}
return advertisement
def _build_daemon_health_record(self, pulse: RepixelizerRuntimePulse) -> dict[str, Any]:
queue = pulse.queue
auth_mode = pulse.auth_payload.get("mode") or "off"
health_transport = "CultNet/RUDP" if self.runtime_config.idunn_rudp_health else "CultCache witness only"
active = queue["activeStatus"] or "idle"
detail = (
f"Repixelizer GUI runtime active; hostedDemo={self.hosted_config.hosted_demo}; "
f"queue={queue['queueDepth']}/{queue['queueCapacity']} waiting={queue['waitingCount']}; "
f"active={active}; auth={auth_mode}; healthTransport={health_transport}"
)
return {
"daemon_id": self.runtime_config.daemon_id,
"state": "active",
"detail": detail,
"observed_at": pulse.updated_at,
"health_contract": self.runtime_config.idunn_health_contract,
"publication_source": "daemon-published" if self.runtime_config.idunn_rudp_health else "daemon-published-cultcache",
"transport": CULTNET_RUDP_PROTOCOL_ID if self.runtime_config.idunn_rudp_health else "cultcache-store",
}
def _publish_idunn_health(self, key: str, stored_at: str, payload: bytes) -> None:
endpoint = self.runtime_config.idunn_rudp_health
if not endpoint:
return
bindings = _cultlib()
client = bindings.CultMesh.create_rudp_client(
runtime_id=self.runtime_config.daemon_id,
connection_id=IDUNN_HEALTH_RUDP_CONNECTION_ID,
endpoint=f"rudp://{endpoint}",
bind_host="0.0.0.0",
)
try:
client.socket.settimeout(0.2)
client.connect()
deadline = time.time() + 2.0
while time.time() < deadline and not client.connected:
client.poll_resends()
client.receive_once()
time.sleep(0.02)
if not client.connected:
raise RuntimeError(f"Idunn RUDP health connect timed out for {endpoint}")
message = bindings.document_put_raw(
message_id=f"{self.runtime_config.daemon_id}-{int(time.time() * 1000)}",
key=key,
schema_id="idunn.daemon_health",
stored_at=stored_at,
payload=payload,
source_runtime_id=self.runtime_config.daemon_id,
)
client.send(IDUNN_HEALTH_CHANNEL_ID, message.to_bytes())
settle_deadline = time.time() + 1.0
while time.time() < settle_deadline:
client.poll_resends()
client.receive_once()
time.sleep(0.02)
finally:
with contextlib.suppress(Exception):
client.disconnect()
client.close()
def _current_transport(self) -> str:
if self.runtime_config.idunn_rudp_health:
return "daemon-owned-cultcache-witness + daemon-published-rudp-health + compatibility.ssh-systemd-http"
return "daemon-owned-cultcache-witness + compatibility.ssh-systemd-http"
def _transport_cut_line(self) -> str:
if self.runtime_config.idunn_rudp_health:
return "Repixelizer now publishes daemon-owned witness state and Idunn health from the live GUI runtime; systemd, /api/health, /api/config, and nginx routing are deployment/debug witnesses only once Odin and Idunn consume the typed surfaces."
return "Repixelizer now publishes a daemon-owned witness store from the live GUI runtime; the remaining transport debt is Idunn health publication over CultNet/RUDP."
def _display_witness_path(workspace_root: Path, cultcache_path: Path) -> str:
with contextlib.suppress(ValueError):
return str(cultcache_path.relative_to(workspace_root)).replace("\\", "/")
with contextlib.suppress(ValueError):
return os.path.relpath(cultcache_path, workspace_root).replace("\\", "/")
return str(cultcache_path)
def _summarize_event_payload(payload: dict[str, Any]) -> dict[str, Any]:
summary: dict[str, Any] = {}
for key, value in payload.items():
if key in IMAGEISH_KEYS or "Image" in key or key.endswith("Image"):
continue
if isinstance(value, (str, int, float, bool)) or value is None:
summary[key] = value
continue
if isinstance(value, list):
scalar_values = [item for item in value[:MAX_EVENT_LIST_ITEMS] if isinstance(item, (str, int, float, bool)) or item is None]
if scalar_values:
summary[key] = scalar_values
continue
if isinstance(value, dict):
compact: dict[str, Any] = {}
for inner_key, inner_value in list(value.items())[:MAX_EVENT_DICT_ITEMS]:
if inner_key in IMAGEISH_KEYS or "Image" in inner_key or inner_key.endswith("Image"):
continue
if isinstance(inner_value, (str, int, float, bool)) or inner_value is None:
compact[inner_key] = inner_value
if compact:
summary[key] = compact
return summary

9
src/repixelizer/witness.py
View file

@ -113,12 +113,14 @@ def build_provider_advertisement(
{
"surfaceId": "repixelizer.operator.surface",
"schema": "gamecult.eve.surface.v1",
"transport": "cultmesh-document",
"transport": "http-json",
"key": eve_surface_key,
"url": f"{public_base_url.rstrip('/')}/eve/surface",
"audience": "operator",
"mode": "read-only-first-cut",
"commands": [],
"status": "planned",
"commands": ["repixelizer.web.open", "repixelizer.job.submit", "repixelizer.job.cancel_own"],
"status": "available",
"styleProfile": "repixelizer.retro.pixel",
},
{
"surfaceId": "repixelizer.web.app",
@ -179,6 +181,7 @@ def build_provider_advertisement(
{
"styleProfile": "repixelizer.product",
"tokenGroups": [
"repixelizer.retro.pixel",
"repixelizer.comparisonCanvas",
"repixelizer.cleanupTool",
"repixelizer.queuePanel",

1
state/evidence.jsonl
View file

@ -35,3 +35,4 @@
{"ts": "2026-05-30T23:39:55+00:00", "type": "implementation", "status": "ok", "note": "Spritesheet automatic sizing now uses a sheet-level median texel density before reconstruction; on the 12 generated spritesheets in 019e7aa3-4bbb-7302-bafd-273f2aa31e53, worst per-sheet density drift improved from 42.21% to 1.786% across 153 sprites."}
{"ts": "2026-05-31T01:13:17+00:00", "type": "implementation", "status": "ok", "note": "Spritesheet mode now supports explicit grid ownership via sheet_columns/sheet_rows plus individual sprite export. The 12 generated character sheets were rerun as 3x2 grids, producing 72 exports with no component chopping and worst within-sheet density deviation 0.124%."}
{"ts": "2026-06-03T02:21:02+00:00", "type": "implementation", "status": "ok", "note": "Added read-only Repixelizer provider advertisement fixture/export naming job/queue/event/runtime/auth witness schemas and reserved state/repixelizer.witness.cc path without moving gui.py queue ownership."}
{"ts": "2026-06-04T01:55:00+00:00", "type": "implementation", "status": "ok", "note": "Added read-only Repixelizer Eve surface projection: CLI fixture/export, hosted /eve/surface route, retro pixel style tokens, route-owned command descriptors, and Android/Periwinkle lowering comparison without moving gui.py queue ownership."}

28
state/map.yaml
View file

@ -54,6 +54,7 @@ accepted_design:
- `src/repixelizer/access.py` owns the hosted access seam: trusted-header ingress still works, `GC_ACCESS_MODE=heimdall` verifies Heimdall Ed25519 access tokens locally, and auth attempts are adopted into local httpOnly cookie sessions.
- In Heimdall mode, Repixelizer keeps a separate httpOnly refresh cookie and asks Heimdall to reissue a short-lived access claim before forcing the user back through provider OAuth.
- `src/repixelizer/witness.py` owns the read-only first-cut Odin provider advertisement export. It names Repixelizer job, queue, event, runtime, and auth witness schemas plus the reserved `state/repixelizer.witness.cc` path, but it does not own queue truth or write CultCache.
- `src/repixelizer/eve_surface.py` owns the read-only Eve surface projection for Repixelizer's front page, single-page workflow, runtime/auth/queue status, retro style tokens, command descriptors, and Periwinkle lowering comparison; `/eve/surface` serves the live projection from `gui.py` without owning queue truth.
- The solver starts from fixed lattice centers, adds tiny seeded noise, follows a direct hierarchical flatness signal, enforces local lattice coherence through spring relaxation, then rounds to real source pixels for final output.
- Cleanup is local and secondary; it does not define the core reconstruction model.
- `source_fidelity` and `source_structure` separate lattice agreement from visible structure.
@ -70,6 +71,7 @@ landed_baseline:
- `source_structure` reporting alongside `source_fidelity`.
- Hosted landing page and Heimdall auth seam in the web layer.
- Read-only `gamecult.eve.provider_advertisement.v1` fixture/export for Odin discovery and future CultCache/Eve witness work.
- Read-only `gamecult.eve.surface.v1` fixture/export plus hosted `/eve/surface` projection, preserving Repixelizer's retro CSS identity as Eve style tokens and documenting Android/Periwinkle style-control parity gaps.
rejected_paths:
- name: tray_optimizer_sprawl
@ -101,7 +103,15 @@ current_status:
`gamecult.eve.provider_advertisement.v1` export and checked-in fixture that
names the Repixelizer job/queue/event/runtime/auth witness schemas and
reserved `state/repixelizer.witness.cc` path for Odin/Huginn discovery,
without moving queue ownership out of `gui.py`. Self-host remains
and `src/repixelizer/eve_surface.py` emits a read-only
`gamecult.eve.surface.v1` projection served at `/eve/surface` and checked in
as `docs/fixtures/repixelizer.eve-surface.json`. The Eve surface keeps the
browser app as the lossless retro lowering, publishes Press Start 2P/VT323
and pixel-shell style tokens, routes commands back to existing GUI routes,
and records that Periwinkle's Kotlin renderer can traverse the surface tree
but does not yet honor provider style tokens, pixel fonts, image assets, or
native upload/canvas controls. These exports do not move queue ownership out
of `gui.py`. Self-host remains
permissive by default unless `GC_ACCESS_*` turns the seam on. The solver is
shipping with a usable but not final tuned config: local search radius
`0.215`, blend `0.24`, move weight `0.09`, grid alignment `4.75`, signal
@ -112,8 +122,16 @@ current_status:
plus stronger blend is the current motion-geometry basin. It did not fully
retune signal knobs around that basin.
next_action: >
Ship the hosted experiment with the current usable solver config so
Repixelizer stops blocking Heimdall and StreamPixels. Later tuning should
start from `docs/solver-tuning-lessons-2026-04-29.md`: sweep signal knobs
around `radius=0.215, blend=0.24` first, then movement/coherence knobs, then
Repixelizer is now the next Idunn transport cut on Yggdrasil. Keep the
current solver config stable, but move daemon truth out of
`repixelizer-gui.service` plus `/api/health` / `/api/config` compatibility
checks by wiring the existing Python CultLib lane
(`E:\Projects\CultLib\packages\cultcache-py`) into the live GUI runtime.
The next implementation pass should add a daemon-owned `.cc` witness export
from `GuiJobManager`, publish Repixelizer daemon health to Idunn over
CultNet/RUDP, publish runtime-owned command-boundary and transport-profile
state, and teach the Yggdrasil deploy lane to ship the CultLib Python
snapshot alongside the app artifact. Later tuning should start from
`docs/solver-tuning-lessons-2026-04-29.md`: sweep signal knobs around
`radius=0.215, blend=0.24` first, then movement/coherence knobs, then
combinations, and only then run final full-corpus validation.

64
tests/test_eve_surface.py Normal file
View file

@ -0,0 +1,64 @@
from __future__ import annotations
import json
from pathlib import Path
from repixelizer.cli import main as cli_main
from repixelizer.eve_surface import build_repixelizer_eve_surface
def test_repixelizer_eve_surface_preserves_retro_style_tokens(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setenv("REPIXELIZER_HOSTED_DEMO", "1")
monkeypatch.setenv("REPIXELIZER_QUEUE_CAPACITY", "7")
monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(tmp_path / "spool"))
monkeypatch.setenv("GC_ACCESS_MODE", "off")
payload = build_repixelizer_eve_surface(
updated_at="2026-06-04T00:00:00Z",
public_base_url="https://repixelizer.example",
)
assert payload["schema"] == "gamecult.eve.surface.v1"
assert payload["providerId"] == "repixelizer"
assert payload["surface"]["root"]["props"]["styleProfile"] == "repixelizer.retro.pixel"
assert payload["runtime"]["hostedDemo"] is True
assert payload["runtime"]["queue"]["queueCapacity"] == 7
tokens = payload["surface"]["styles"]["tokens"]
assert "Press Start 2P" in tokens["fontTitle"]
assert "VT323" in tokens["fontBody"]
assert tokens["colorBackgroundTop"] == "#01040b"
assert tokens["colorAccent"] == "#ffd84a"
assert tokens["imageRendering"] == "pixelated"
assert tokens["scanlineOverlay"] is True
commands = {entry["command"]: entry for entry in payload["commands"]}
assert commands["repixelizer.job.submit"]["route"] == "/api/jobs"
assert commands["repixelizer.job.cancel_own"]["method"] == "DELETE"
android = payload["loweringComparison"]["periwinkleAndroidKotlin"]
assert android["status"] == "surface-tree-renderer-present-style-token-gap"
assert any("Press Start 2P" in gap for gap in android["losesUntilRendererParity"])
def test_eve_surface_cli_writes_json(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(tmp_path / "spool"))
monkeypatch.setenv("GC_ACCESS_MODE", "off")
out = tmp_path / "repixelizer.eve-surface.json"
exit_code = cli_main(
[
"eve-surface",
"--out",
str(out),
"--updated-at",
"2026-06-04T00:00:00Z",
"--public-base-url",
"https://repixelizer.example",
]
)
assert exit_code == 0
payload = json.loads(out.read_text(encoding="utf-8"))
assert payload["updatedAt"] == "2026-06-04T00:00:00Z"
assert payload["surface"]["styles"]["assets"]["appUrl"] == "https://repixelizer.example/app/"

20
tests/test_gui.py
View file

@ -1195,6 +1195,26 @@ def test_gui_queue_panel_defaults_off_for_local_runs_and_can_be_forced(monkeypat
assert payload["ui"]["showQueuePanel"] is True
def test_gui_serves_eve_surface_from_live_runtime_projection(monkeypatch, tmp_path: Path) -> None:
from repixelizer.gui import create_app
monkeypatch.setenv("REPIXELIZER_HOSTED_DEMO", "1")
monkeypatch.setenv("REPIXELIZER_QUEUE_CAPACITY", "5")
monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(tmp_path / "spool"))
monkeypatch.setenv("GC_ACCESS_MODE", "off")
app = create_app()
status, _headers, body = asyncio.run(_get_response(app, "/eve/surface"))
payload = _response_json(type("Response", (), {"body": body})())
assert status == 200
assert payload["schema"] == "gamecult.eve.surface.v1"
assert payload["surface"]["styles"]["tokens"]["colorAccent"] == "#ffd84a"
assert payload["runtime"]["queue"]["queueCapacity"] == 5
assert payload["runtime"]["auth"]["mode"] == "off"
assert payload["surface"]["styles"]["assets"]["appUrl"] == "http://testserver/app/"
def test_gui_queue_rejects_eleventh_waiting_job(monkeypatch, tmp_path: Path) -> None:
from repixelizer.gui import create_app

49
tests/test_verse_state.py Normal file
View file

@ -0,0 +1,49 @@
from __future__ import annotations
from pathlib import Path
from repixelizer.access import AccessController
from repixelizer.gui import GuiJobManager, HostedDemoConfig
from repixelizer.verse_state import RepixelizerVerseRuntime, RepixelizerVerseRuntimeConfig
def test_gui_module_imports_cleanly() -> None:
import repixelizer.gui as gui
assert callable(gui.create_app)
def test_verse_runtime_publish_once_writes_witness(tmp_path: Path, monkeypatch) -> None:
spool_dir = tmp_path / "spool"
witness_path = tmp_path / "cultcache" / "repixelizer.service.cc"
monkeypatch.setenv("REPIXELIZER_VERSE_RUNTIME", "1")
monkeypatch.setenv("REPIXELIZER_SPOOL_DIR", str(spool_dir))
monkeypatch.setenv("GC_ACCESS_MODE", "off")
monkeypatch.setenv("GC_ACCESS_REQUIRED", "0")
monkeypatch.setenv("GC_ACCESS_CULTCACHE_PATH", str(witness_path))
monkeypatch.setenv("GC_ACCESS_IDUNN_DAEMON", "repixelizer-test")
monkeypatch.setenv("GC_ACCESS_IDUNN_HEALTH_CONTRACT", "repixelizer.cultnet-rudp-service-health")
monkeypatch.delenv("GC_ACCESS_IDUNN_RUDP_HEALTH", raising=False)
config = HostedDemoConfig.from_env()
manager = GuiJobManager(config)
access_controller = AccessController.from_env(hosted_demo=config.hosted_demo)
runtime = RepixelizerVerseRuntime(
RepixelizerVerseRuntimeConfig.from_env(config),
config,
access_controller,
manager,
)
runtime.publish_once()
health = runtime.build_health_payload()
assert runtime.enabled is True
assert witness_path.exists()
assert witness_path.stat().st_size > 0
assert health["status"] == "ok"
assert health["verseRuntime"] == "enabled"
assert health["cultCachePath"] == str(witness_path)
assert health["idunnDaemon"] == "repixelizer-test"
assert health["idunnRudpHealth"] is None

5
tests/test_witness.py
View file

@ -33,6 +33,11 @@ def test_provider_advertisement_names_witness_schemas(monkeypatch, tmp_path: Pat
assert witness["kind"] == "cc-export-path-reserved"
assert witness["path"] == "state/repixelizer.witness.cc"
assert witness["freshness"]["state"] == "planned"
surface = payload["surfaces"][0]
assert surface["status"] == "available"
assert surface["transport"] == "http-json"
assert surface["url"] == "https://repixelizer.gamecult.org/eve/surface"
assert surface["styleProfile"] == "repixelizer.retro.pixel"
def test_witness_advertisement_cli_writes_json(monkeypatch, tmp_path: Path) -> None: