Dwarf Society — Runtime Boundary Implementation Plan

$DfRoot = 'D:\SteamLibrary\steamapps\common\Dwarf Fortress'
$stamp = Get-Date -AsUTC -Format 'yyyyMMddTHHmmssZ'
Select-String -Path "$DfRoot\release notes.txt" -Pattern '^Release notes for 53\.15'
if (Test-Path "$DfRoot\hack") { throw 'Unexpected DFHack install already present' }
$backup = "D:\SteamLibrary\df-saves-backup-$stamp"
New-Item -ItemType Directory -Path $backup | Out-Null
robocopy "$DfRoot\save" "$backup\save" /E /COPY:DAT /DCOPY:DAT /R:1 /W:1
if ($LASTEXITCODE -gt 7) { throw "save backup failed: $LASTEXITCODE" }
Get-FileHash "$DfRoot\Dwarf Fortress.exe" -Algorithm SHA256

Set-Location C:\Users\parzi\source\repos
gh repo create dkta-labs/dwarf-society --private --clone
Set-Location .\dwarf-society
uv init --package --name dwarf-society --python 3.13
Remove-Item .\README.md -ErrorAction SilentlyContinue
uv add 'pydantic>=2.11,<3' 'httpx>=0.28,<1'
uv add --dev 'pytest>=8.4,<9' 'ruff>=0.12,<1'
New-Item -ItemType Directory -Force -Path `
  src\dwarf_society,tests\unit,tests\integration,tests\fixtures\runtime, `
  dfhack\scripts\dwarf-society,scripts,runtime | Out-Null

.venv/
__pycache__/
.pytest_cache/
.ruff_cache/
*.pyc
runtime/*
!runtime/.gitkeep

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-ra"
markers = ["live: requires a running disposable DFHack fortress"]

[tool.ruff]
target-version = "py313"
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]

import dwarf_society


def test_package_imports() -> None:
    assert dwarf_society.__package__ == "dwarf_society"

git add .gitignore .python-version pyproject.toml uv.lock runtime/.gitkeep `
  src/dwarf_society/__init__.py tests/unit/test_smoke.py
git commit -m "chore: bootstrap dwarf society runtime"

from uuid import uuid4

import pytest
from pydantic import ValidationError

from dwarf_society.contracts import PROTOCOL_VERSION, RequestEnvelope, ResponseEnvelope


def test_request_rejects_unknown_fields() -> None:
    payload = {
        "protocol": PROTOCOL_VERSION,
        "request_id": str(uuid4()),
        "kind": "health",
        "payload": {},
        "unexpected": True,
    }
    with pytest.raises(ValidationError):
        RequestEnvelope.model_validate(payload)


def test_response_requires_matching_protocol() -> None:
    with pytest.raises(ValidationError):
        ResponseEnvelope.model_validate({
            "protocol": "999",
            "request_id": str(uuid4()),
            "status": "ok",
            "result": {},
        })

from typing import Literal
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field, model_validator

PROTOCOL_VERSION = "1"


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)


class RequestEnvelope(StrictModel):
    protocol: Literal["1"] = PROTOCOL_VERSION
    request_id: UUID
    kind: Literal["health", "observe", "advance", "claim_job"]
    run_id: UUID | None = None
    save_id: str | None = None
    bridge_epoch: str | None = None
    expected_revision: int | None = Field(default=None, ge=0)
    command_id: UUID | None = None
    payload: dict[str, object]


class ProtocolError(StrictModel):
    code: Literal[
        "invalid_request", "wrong_save", "wrong_epoch", "stale_revision", "bridge_busy",
        "unsupported_capability", "precondition_failed", "internal_error"
    ]
    message: str


class ResponseEnvelope(StrictModel):
    protocol: Literal["1"] = PROTOCOL_VERSION
    request_id: UUID
    status: Literal["ok", "rejected", "error"]
    save_id: str | None = None
    bridge_epoch: str | None = None
    observed_revision: int | None = Field(default=None, ge=0)
    result: dict[str, object] | None = None
    error: ProtocolError | None = None

    @model_validator(mode="after")
    def error_matches_status(self):
        if self.status == "ok" and self.error is not None:
            raise ValueError("ok response cannot carry an error")
        if self.status != "ok" and self.error is None:
            raise ValueError("non-ok response requires an error")
        return self

git add src/dwarf_society/contracts.py tests/unit/test_contracts.py
git commit -m "feat: define strict bridge contracts"

class BridgeTimeout(TimeoutError):
    pass


class SpoolClient:
    def __init__(self, root: Path) -> None:
        self.root = root
        self.requests = root / "requests"
        self.responses = root / "responses"
        self.archive = root / "archive"
        for path in (self.requests, self.responses, self.archive):
            path.mkdir(parents=True, exist_ok=True)

    @staticmethod
    def _write_atomic(path: Path, content: str) -> None:
        temp = path.with_suffix(path.suffix + ".tmp")
        with temp.open("w", encoding="utf-8", newline="\n") as handle:
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp, path)

    def request(self, envelope: RequestEnvelope, timeout_s: float = 10.0) -> ResponseEnvelope:
        name = f"{time.time_ns():020d}-{envelope.request_id}.json"
        request_path = self.requests / name
        self._write_atomic(request_path, envelope.model_dump_json())
        response_path = self.responses / f"{envelope.request_id}.json"
        deadline = time.monotonic() + timeout_s
        while time.monotonic() < deadline:
            if response_path.exists():
                response = ResponseEnvelope.model_validate_json(response_path.read_text("utf-8"))
                if response.request_id != envelope.request_id:
                    raise ValueError("response request_id mismatch")
                os.replace(response_path, self.archive / response_path.name)
                return response
            time.sleep(0.025)
        raise BridgeTimeout(f"bridge timed out after {timeout_s:.3f}s")

local function write_response(request_id, response)
    local final = paths.responses .. request_id .. '.json'
    local temp = final .. '.tmp'
    json.encode_file(response, temp, {pretty=false})
    assert(os.rename(temp, final))
end

--@ enable=true
--@ module=true
local repeat_util = require('repeat-util')
local protocol = reqscript('dwarf-society/protocol')
local KEY = 'dwarf-society'
state = state or {enabled=false}
runtime = runtime or {}

local function poll()
    protocol.process_one()
end

local function do_enable()
    runtime.bridge_epoch =
        ('%d-%d-%s'):format(os.time(), dfhack.getTickCount(), tostring({}))
    protocol.init_spool(runtime)
    repeat_util.scheduleEvery(KEY, 5, 'frames', poll)
end

local function do_disable()
    repeat_util.cancel(KEY)
end

function isEnabled() return state.enabled end
function setEnabled(enabled)
    if enabled == state.enabled then return end
    state.enabled = enabled
    if enabled then do_enable() else do_disable() end
end

git add src/dwarf_society/ipc.py tests/unit/test_ipc.py dfhack scripts/install_bridge.ps1
git commit -m "feat: add paused-safe bridge transport"

local function enum_name(enum, value)
    return enum[value] or ('UNKNOWN_' .. tostring(value))
end

local function extract_needs(personality)
    local out = {}
    for _, need in ipairs(personality.needs) do
        table.insert(out, {
            name=enum_name(df.need_type, need.id),
            deity_id=need.deity_id,
            focus_level=need.focus_level,
            need_level=need.need_level,
        })
    end
    return out
end

local function extract_skills(soul)
    local out = {}
    for _, skill in ipairs(soul.skills) do
        table.insert(out, {
            name=enum_name(df.job_skill, skill.id),
            rating=skill.rating,
            experience=skill.experience,
        })
    end
    return out
end

uv run pytest tests/unit tests/integration/test_runtime_boundary.py -v
uv run ruff check .

git add dfhack/scripts/dwarf-society/snapshot.lua \
  src/dwarf_society/contracts.py src/dwarf_society/client.py \
  tests/integration/test_runtime_boundary.py tests/fixtures/runtime/initial_snapshot.json
git commit -m "feat: expose canonical citizen snapshot"

local function advance(request, respond)
    validate_mutation_fence(request)
    if runtime.busy then
        return reject(request, 'bridge_busy', 'advance already pending')
    end
    local ticks = tonumber(request.payload.ticks)
    if not ticks or ticks < 1 or ticks > 1200 then
        return reject(request, 'invalid_request', 'ticks must be 1..1200')
    end
    if not dfhack.world.ReadPauseState() then
        return reject(request, 'precondition_failed', 'game must begin paused')
    end
    runtime.busy = {request_id=request.request_id, kind='advance'}
    dfhack.timeout(ticks, 'ticks', function()
        dfhack.world.SetPauseState(true)
        state.revision = state.revision + 1
        persist_state()
        respond(ok_receipt(request, {ticks=ticks, busy=false}))
        runtime.busy = nil
    end)
    dfhack.world.SetPauseState(false)
    return 'deferred'
end

git add dfhack src/dwarf_society tests
git commit -m "feat: add fenced pause-step clock"

{
  command_id = request.command_id,
  capability = 'claim_job',
  outcome = 'applied',
  unit_id = unit.id,
  job_id = job.id,
  before_worker_unit_id = -1,
  after_worker_unit_id = unit.id,
  revision_before = revision_before,
  revision_after = state.revision,
}

git add dfhack src/dwarf_society tests
git commit -m "feat: add idempotent legal job claim"

Invoke-RestMethod -Method Get -TimeoutSec 5 `
  -Uri 'http://dakotas-mac-studio:8080/v1/models'

ssh.exe -NT -L 127.0.0.1:18080:127.0.0.1:8080 dakotas-mac-studio

uv run pytest -m "not live" -v
$env:DWARF_SOCIETY_LIVE = '1'
uv run pytest -m live tests/integration/test_runtime_boundary.py -v
uv run ruff check .
uv run python -m dwarf_society.cli report-runtime-boundary

git add README.md src/dwarf_society tests pyproject.toml uv.lock
git commit -m "docs: record runtime boundary proof"