palamedes/palamedes_reference_adapter.py at main · LEE-Kyungjae/palamedes · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
#!/usr/bin/env python3
from dataclasses import dataclass
from typing import Any, Dict

from palamedes_sdk import (
    PalamedesClient,
    PalamedesClientOperationError,
    PalamedesConflictError,
    PalamedesHealthGateError,
)


@dataclass
class PalamedesKernelAdapter:
    client: PalamedesClient
    history_limit: int = 5
    require_healthy_writes: bool = True

    def snapshot(self) -> Dict[str, Any]:
        return self.client.get_cycle(history_limit=self.history_limit)

    def apply_plan_update(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        return self.client.apply_and_get_cycle_with_retry(
            "update_plan",
            payload,
            history_limit=self.history_limit,
            require_healthy=self.require_healthy_writes,
        )

    def capture_evidence(self, payload: Dict[str, Any], *, allow_retry: bool = False) -> Dict[str, Any]:
        return self.client.apply_and_get_cycle_with_retry(
            "add_evidence",
            payload,
            history_limit=self.history_limit,
            require_healthy=self.require_healthy_writes,
            allow_non_idempotent_retry=allow_retry,
        )

    def restore_previous(self) -> Dict[str, Any]:
        return self.client.apply_and_get_cycle_with_retry(
            "restore_revision",
            {"previous": True},
            history_limit=self.history_limit,
            require_healthy=self.require_healthy_writes,
        )

    def preview_restore_previous(self) -> Dict[str, Any]:
        return self.client.preview_restore(previous=True)


def summarize_for_host(result: Dict[str, Any]) -> Dict[str, Any]:
    post_cycle = result.get("post_cycle", {})
    qa = post_cycle.get("qa", {})
    health = post_cycle.get("health", {})
    return {
        "operation": result.get("operation", ""),
        "fingerprint": result.get("post_fingerprint", ""),
        "changed_fields": result.get("changed_fields", []),
        "qa_result": qa.get("result", ""),
        "qa_score": qa.get("score"),
        "health_status": health.get("status", ""),
        "retried": bool(result.get("retried", False)),
    }


def example_host_step() -> None:
    adapter = PalamedesKernelAdapter(PalamedesClient.from_http("127.0.0.1", 8787))
    try:
        cycle = adapter.snapshot()
        print("Current goal:", cycle["plan"].get("goal", ""))
        result = adapter.apply_plan_update(
            {
                "goal": "Narrow to creator workflow automation",
                "success_metric": "Reach 5 retained pilots",
                "deadline": "2026-04-30",
            }
        )
        print(summarize_for_host(result))
    except PalamedesConflictError as exc:
        print(
            {
                "type": "conflict",
                "operation": exc.operation,
                "step": exc.step,
                "expected_fingerprint": exc.expected_fingerprint,
                "current_fingerprint": exc.current_fingerprint,
            }
        )
    except PalamedesHealthGateError as exc:
        print({"type": "health_gate", "operation": exc.operation, "status": exc.status})
    except PalamedesClientOperationError as exc:
        print({"type": "operation_error", "operation": exc.operation, "step": exc.step, "status": exc.status})


if __name__ == "__main__":
    example_host_step()