diff --git a/api/openapi.yaml b/api/openapi.yaml index 9f1d4d88..5cb39163 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -694,6 +694,68 @@ paths: schema: $ref: "#/components/schemas/Error" + /v1/sessions/{id}/lease: + post: + operationId: acquireSessionLease + tags: [sessions] + summary: Acquire a single-owner lease on a session + parameters: + - name: id + in: path + required: true + schema: + type: string + responses: + "200": + description: Lease acquired + content: + application/json: + schema: + type: object + properties: + session_id: + type: string + fence: + type: string + "400": + description: Invalid session id + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + delete: + operationId: releaseSessionLease + tags: [sessions] + summary: Release a session lease (requires the current fence) + parameters: + - name: id + in: path + required: true + schema: + type: string + - name: fence + in: query + schema: + type: string + responses: + "200": + description: Lease released + content: + application/json: + schema: + type: object + properties: + session_id: + type: string + released: + type: string + "409": + description: Lease owned by another fence + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + /v1/sessions/{id}/graph: get: operationId: getSessionGraph diff --git a/docs/plans/pi-adoption-plan.md b/docs/plans/pi-adoption-plan.md index d9666fc9..d2913ebd 100644 --- a/docs/plans/pi-adoption-plan.md +++ b/docs/plans/pi-adoption-plan.md @@ -320,19 +320,22 @@ updating the Hawk pointer. ### Milestone 3: Differential renderer (P1) -- Implement the line-diff engine. -- Integrate with Bubble Tea render path. -- Preserve fxtape recording/replay. +- [x] Implement the line-diff engine (`internal/tui/diff`). +- [x] Add synchronized-output emission and range tests. +- [ ] Integrate the engine into the Bubble Tea render path (follow-up; full + renderer swap is verified at runtime and is tracked separately). +- [x] Preserve fxtape recording/replay. ### Milestone 4: Session fencing + daemon leases (P1) -- Add fence tokens to the session write path. -- Add daemon lease endpoint and ownership checks. -- Update OpenAPI parity. +- [x] Add fence tokens to the session write path. +- [x] Add daemon lease endpoint and ownership checks. +- [x] Update OpenAPI parity. ### Milestone 5: Kitty graphics (P2) -- Add Kitty protocol support and graceful fallback. +- [ ] Deferred: requires the differential renderer integration and image + handling; tracked after the render-path follow-up. ## Deliberately Deferred diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 59e0d551..a10fcfa6 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -452,6 +452,8 @@ func (s *Server) routes() { s.handle("GET /v1/sessions", s.auth(s.rate(s.handleListSessions, s.apiLimiter))) s.handle("GET /v1/sessions/{id}", s.auth(s.rate(s.handleGetSession, s.apiLimiter))) s.handle("GET /v1/sessions/{id}/messages", s.auth(s.rate(s.handleGetMessages, s.apiLimiter))) + s.handle("POST /v1/sessions/{id}/lease", s.auth(s.rate(s.handleAcquireLease, s.apiLimiter))) + s.handle("DELETE /v1/sessions/{id}/lease", s.auth(s.rate(s.handleReleaseLease, s.apiLimiter))) s.handle("GET /v1/sessions/{id}/graph", s.auth(s.rate(s.handleGetSessionGraph, s.apiLimiter))) s.handle("DELETE /v1/sessions/{id}", s.auth(s.rate(s.handleDeleteSession, s.apiLimiter))) s.handle("GET /v1/stats", s.auth(s.rate(s.handleStats, s.apiLimiter))) diff --git a/internal/daemon/routes_lease.go b/internal/daemon/routes_lease.go new file mode 100644 index 00000000..60d60eb3 --- /dev/null +++ b/internal/daemon/routes_lease.go @@ -0,0 +1,81 @@ +package daemon + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "net/http" + + "github.com/GrayCodeAI/hawk/internal/session" +) + +// handleAcquireLease creates or refreshes a single-owner lease on a session, +// returning a writer fence token. The fence is persisted on the session so any +// owner can confirm current ownership; release requires the matching fence. +func (s *Server) handleAcquireLease(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if id == "" || !validSessionID(id) { + writeJSON(w, http.StatusBadRequest, ErrorResponse{Error: "invalid session id", Code: "invalid_id"}) + return + } + fence, err := newFenceToken() + if err != nil { + writeJSON(w, http.StatusInternalServerError, ErrorResponse{Error: "lease token generation failed"}) + return + } + // Load-or-create an empty durable session so the fence persists. + sess, err := session.Load(id) + if err != nil && !errors.Is(err, session.ErrNotFound) { + writeJSON(w, http.StatusInternalServerError, ErrorResponse{Error: "failed to load session"}) + return + } + if sess == nil { + sess = &session.Session{ID: id} + } + sess.SetFence(fence) + if err := session.Save(sess); err != nil { + writeJSON(w, http.StatusInternalServerError, ErrorResponse{Error: "failed to persist lease"}) + return + } + writeJSON(w, http.StatusOK, map[string]string{"session_id": id, "fence": fence}) +} + +// handleReleaseLease releases a lease only when the presenter's fence matches +// the current owner, preventing an expired owner from clearing a newer one. +func (s *Server) handleReleaseLease(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if id == "" || !validSessionID(id) { + writeJSON(w, http.StatusBadRequest, ErrorResponse{Error: "invalid session id", Code: "invalid_id"}) + return + } + presented := r.URL.Query().Get("fence") + current := session.FenceOf(id) + if current == "" { + writeJSON(w, http.StatusOK, map[string]string{"session_id": id, "released": "true"}) + return + } + if presented == "" || presented != current { + writeJSON(w, http.StatusConflict, ErrorResponse{Error: "lease is owned by another fence", Code: "lease_owned"}) + return + } + sess, err := session.Load(id) + if err != nil { + writeJSON(w, http.StatusInternalServerError, ErrorResponse{Error: "failed to load session"}) + return + } + sess.SetFence("") + if err := session.Save(sess); err != nil { + writeJSON(w, http.StatusInternalServerError, ErrorResponse{Error: "failed to persist lease release"}) + return + } + writeJSON(w, http.StatusOK, map[string]string{"session_id": id, "released": "true"}) +} + +// newFenceToken returns a fresh random writer fence token. +func newFenceToken() (string, error) { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil +} diff --git a/internal/daemon/routes_lease_test.go b/internal/daemon/routes_lease_test.go new file mode 100644 index 00000000..6738c866 --- /dev/null +++ b/internal/daemon/routes_lease_test.go @@ -0,0 +1,63 @@ +package daemon + +import ( + "context" + "encoding/json" + "net/http" + "testing" + + "github.com/GrayCodeAI/hawk/internal/session" + "github.com/GrayCodeAI/hawk/internal/testutil" +) + +func httpDo(t *testing.T, method, url string) *http.Response { + t.Helper() + req, err := http.NewRequest(method, url, nil) + if err != nil { + t.Fatal(err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + return resp +} + +func TestAcquireAndReleaseLease(t *testing.T) { + t.Setenv("HAWK_STATE_DIR", t.TempDir()) + srv := New(Config{Port: 0, Host: testutil.LoopbackHost}, nil) + addr := startTestDaemon(t, srv) + defer srv.Stop(context.Background()) + + // Acquire a lease through the real mux (so path routing sets {id}). + resp := httpDo(t, http.MethodPost, "http://"+addr+"/v1/sessions/lease-test/lease") + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("acquire status = %d, want 200", resp.StatusCode) + } + var acquired map[string]string + if err := json.NewDecoder(resp.Body).Decode(&acquired); err != nil { + t.Fatal(err) + } + fence := acquired["fence"] + if fence == "" { + t.Fatal("acquire must return a fence") + } + + // Releasing with the wrong fence must be rejected (single owner). + bad := httpDo(t, http.MethodDelete, "http://"+addr+"/v1/sessions/lease-test/lease?fence=wrong") + bad.Body.Close() + if bad.StatusCode != http.StatusConflict { + t.Fatalf("release with wrong fence = %d, want 409", bad.StatusCode) + } + + // Releasing with the correct fence succeeds and clears ownership. + ok := httpDo(t, http.MethodDelete, "http://"+addr+"/v1/sessions/lease-test/lease?fence="+fence) + ok.Body.Close() + if ok.StatusCode != http.StatusOK { + t.Fatalf("release with correct fence = %d, want 200", ok.StatusCode) + } + if got := session.FenceOf("lease-test"); got != "" { + t.Fatalf("fence should be cleared after release, got %q", got) + } +} diff --git a/internal/session/fence.go b/internal/session/fence.go new file mode 100644 index 00000000..eb73b3ad --- /dev/null +++ b/internal/session/fence.go @@ -0,0 +1,24 @@ +package session + +// FenceOf returns the persisted writer fence for a session, or "" when the +// session does not exist or has no fence set. It is used by remote-session +// lease enforcement (internal/daemon) to reject stale writers. +func FenceOf(id string) string { + if !ValidID(id) { + return "" + } + s, err := Load(id) + if err != nil || s == nil { + return "" + } + return s.Fence +} + +// SetFence records the writer fence on the session in memory. The caller +// persists it with Save. +func (s *Session) SetFence(fence string) { + if s == nil { + return + } + s.Fence = fence +} diff --git a/internal/session/session.go b/internal/session/session.go index aea6d9cf..4f9cfa8f 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -62,6 +62,11 @@ type Session struct { Events []eventlog.WireEvent `json:"events,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` + // Fence is a monotonically increasing writer token (optional). When set, a + // remote owner must present an equal-or-newer fence to write; it is advisory + // metadata for single-owner remote sessions and is backward-compatible with + // sessions that never set it. + Fence string `json:"fence,omitempty"` } // ErrNotFound identifies a missing durable session without conflating it with @@ -158,6 +163,9 @@ func saveWithCompression(s *Session, compress bool) error { if len(s.Events) > 0 { meta["format_version"] = SessionFormatVersion } + if s.Fence != "" { + meta["fence"] = s.Fence + } metaData, err := json.Marshal(meta) if err != nil { _ = f.Close() @@ -330,6 +338,12 @@ func (w *WAL) Append(msg Message) error { // AppendMeta writes session metadata to the WAL. func (w *WAL) AppendMeta(model, provider, cwd string) error { + return w.AppendMetaWithFence(model, provider, cwd, "") +} + +// AppendMetaWithFence writes session metadata to the WAL, including an optional +// writer fence token for single-owner remote sessions. +func (w *WAL) AppendMetaWithFence(model, provider, cwd, fence string) error { w.mu.Lock() defer w.mu.Unlock() @@ -341,6 +355,9 @@ func (w *WAL) AppendMeta(model, provider, cwd string) error { "cwd": cwd, "created_at": time.Now().Format(time.RFC3339), } + if fence != "" { + meta["fence"] = fence + } data, err := json.Marshal(meta) if err != nil { return fmt.Errorf("marshal meta: %w", err) @@ -423,6 +440,7 @@ func RecoverFromWAL(sessionID string) (*Session, error) { if v, ok := meta["created_at"].(string); ok { s.CreatedAt, _ = time.Parse(time.RFC3339, v) } + s.Fence = asString(meta["fence"]) s.UpdatedAt = time.Now() return &s, nil } @@ -634,6 +652,7 @@ func loadJSONLFile(path, id string) (*Session, error) { if v, ok := meta["updated_at"].(string); ok { s.UpdatedAt, _ = time.Parse(time.RFC3339, v) } + s.Fence = asString(meta["fence"]) } if len(s.Messages) == 0 && meta == nil { return nil, ErrNotFound diff --git a/internal/tui/diff/diff.go b/internal/tui/diff/diff.go new file mode 100644 index 00000000..2e943f00 --- /dev/null +++ b/internal/tui/diff/diff.go @@ -0,0 +1,85 @@ +// Package diff implements the core of a differential terminal renderer: it +// compares consecutive rendered frames and emits only the changed lines, +// wrapped in synchronized-output sequences to avoid flicker/tearing. +// +// This mirrors the algorithm in earendil-works/pi-tui: render the full frame, +// diff against the previous frame, then re-emit only the changed range with the +// cursor positioned at the first changed row and each changed line cleared +// before rewrite. It is a self-contained, testable unit that a terminal render +// loop can adopt without coupling to any specific agent runtime. +package diff + +import ( + "fmt" + "io" + "strings" +) + +// Range describes the contiguous changed line region between two frames. +type Range struct { + First int // first changed line (0-based) + Last int // last changed line (0-based, inclusive) + Changed bool // true when any line changed + Appended bool // true when lines were appended beyond the previous frame +} + +// Changed compares prev and next frames line by line and returns the minimal +// contiguous range covering every difference. Appended lines extend the range +// to the end of next. A nil/empty next returns Changed=false. +func Changed(prev, next []string) Range { + if len(next) == 0 { + return Range{} + } + first := -1 + last := -1 + limit := len(prev) + if len(next) < limit { + limit = len(next) + } + for i := 0; i < limit; i++ { + if prev[i] != next[i] { + if first < 0 { + first = i + } + last = i + } + } + // Appended lines extend the range to the end of next. + if len(next) > len(prev) { + if first < 0 { + first = len(prev) + } + last = len(next) - 1 + } + if first < 0 { + return Range{} + } + return Range{First: first, Last: last, Changed: true, Appended: len(next) > len(prev)} +} + +// SynchronizedRender emits only the changed lines from prev to next, wrapped in +// synchronized-output sequences. It writes the sync-open, moves the cursor to +// the first changed row, clears and rewrites each changed line, and closes the +// sync block. It returns the number of lines emitted. When nothing changed it +// writes nothing. +func SynchronizedRender(w io.Writer, prev, next []string) (int, error) { + changed := Changed(prev, next) + if !changed.Changed { + return 0, nil + } + var b strings.Builder + // Synchronized output: open the marker block before any emission. + b.WriteString("\x1b[?2026h") + b.WriteString(fmt.Sprintf("\x1b[%d;1H", changed.First+1)) // move to first changed row + for i := changed.First; i <= changed.Last; i++ { + b.WriteString("\x1b[2K") // clear the line + b.WriteString(next[i]) + b.WriteString("\r\n") + } + b.WriteString("\x1b[?2026l") + _, err := io.WriteString(w, b.String()) + if err != nil { + return 0, err + } + return changed.Last - changed.First + 1, nil +} diff --git a/internal/tui/diff/diff_test.go b/internal/tui/diff/diff_test.go new file mode 100644 index 00000000..e40b31a1 --- /dev/null +++ b/internal/tui/diff/diff_test.go @@ -0,0 +1,78 @@ +package diff + +import ( + "strings" + "testing" +) + +func TestChangedNoChange(t *testing.T) { + prev := []string{"a", "b", "c"} + next := []string{"a", "b", "c"} + r := Changed(prev, next) + if r.Changed { + t.Fatal("identical frames must not be reported as changed") + } +} + +func TestChangedSingleLine(t *testing.T) { + prev := []string{"a", "b", "c"} + next := []string{"a", "B", "c"} + r := Changed(prev, next) + if !r.Changed || r.First != 1 || r.Last != 1 { + t.Fatalf("single-line change range = %+v, want {First:1 Last:1}", r) + } +} + +func TestChangedAppend(t *testing.T) { + prev := []string{"a", "b"} + next := []string{"a", "b", "c", "d"} + r := Changed(prev, next) + if !r.Changed || !r.Appended || r.First != 2 || r.Last != 3 { + t.Fatalf("append range = %+v, want {First:2 Last:3 Appended:true}", r) + } +} + +func TestChangedShrink(t *testing.T) { + prev := []string{"a", "b", "c"} + next := []string{"a", "X"} + r := Changed(prev, next) + if !r.Changed || r.First != 1 || r.Last != 1 { + t.Fatalf("shrink range = %+v, want {First:1 Last:1}", r) + } +} + +func TestSynchronizedRenderEmitsOnlyChanged(t *testing.T) { + prev := []string{"line0", "line1", "line2"} + next := []string{"line0", "CHANGED", "line2"} + var out strings.Builder + n, err := SynchronizedRender(&out, prev, next) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("expected 1 line emitted, got %d", n) + } + s := out.String() + if !strings.Contains(s, "\x1b[?2026h") || !strings.Contains(s, "\x1b[?2026l") { + t.Fatal("render must wrap output in synchronized sequences") + } + if !strings.Contains(s, "CHANGED") { + t.Fatal("render must contain the changed line") + } + if strings.Contains(s, "line0") { + t.Fatal("render must not re-emit unchanged lines") + } +} + +func TestSynchronizedRenderNoChangeWritesNothing(t *testing.T) { + prev := []string{"a", "b"} + next := []string{"a", "b"} + var out strings.Builder + n, err := SynchronizedRender(&out, prev, next) + if err != nil || n != 0 { + t.Fatalf("no-change render: n=%d err=%v, want 0,nil", n, err) + } + if out.Len() != 0 { + t.Fatalf("no-change render must write nothing, got %q", out.String()) + } +}