agents: sessions carry the terminal they publish
A live session already records the machine, repo and cwd it runs on. What was missing to WATCH one is its address: the URL that machine published for its terminal (zrok gives one without opening a port). One column, carried through register, patch and the view, alongside the execution context it belongs to. It is a URL rather than a stream because the bytes belong to the machine running the shell — cloud holds the address, never the connection, so a session that ends stops answering in its own frame instead of leaving a console holding a half-open stream. https only: the console frames this value, and any other scheme is a way to get a javascript: or file: URL rendered on a signed-in page. A pointer on patch, so a session that stops sharing can withdraw it. This REPLACES the separate /v1/sessions plane added earlier in this branch, which was a second answer to a question /v1/agents/sessions already answers. Co-authored-by: Hanzo Dev <dev@hanzo.ai>
This commit is contained in:
+46
-5
@@ -60,6 +60,7 @@ const (
|
||||
maxHost = 256
|
||||
maxCwd = 1024
|
||||
maxRepo = 512
|
||||
maxTerminal = 512
|
||||
maxProvider = 64
|
||||
maxAccount = 256
|
||||
)
|
||||
@@ -90,9 +91,12 @@ type sessionView struct {
|
||||
TaskRunID string `json:"taskRunId,omitempty"`
|
||||
// Execution context (mission-control): the machine/repo/cwd a card shows and
|
||||
// the run-target a session is dispatched to. Omitted when a surface didn't report it.
|
||||
Host string `json:"host,omitempty"`
|
||||
Cwd string `json:"cwd,omitempty"`
|
||||
Repo string `json:"repo,omitempty"`
|
||||
Host string `json:"host,omitempty"`
|
||||
Cwd string `json:"cwd,omitempty"`
|
||||
Repo string `json:"repo,omitempty"`
|
||||
// Terminal is where this session can be WATCHED — the URL the machine
|
||||
// published for its live terminal. Omitted when it publishes none.
|
||||
Terminal string `json:"terminal,omitempty"`
|
||||
Target string `json:"target,omitempty"`
|
||||
Provider string `json:"provider,omitempty"`
|
||||
Account string `json:"account,omitempty"`
|
||||
@@ -171,7 +175,7 @@ func toSessionView(x Session, events, children int) sessionView {
|
||||
ID: x.ID, Org: x.Org, Agent: x.Agent, Actor: x.Actor, Status: x.Status,
|
||||
ParentSessionID: x.ParentID, RootSessionID: x.RootID, Title: x.Title,
|
||||
TaskWorkflowID: x.TaskWorkflowID, TaskRunID: x.TaskRunID,
|
||||
Host: x.Host, Cwd: x.Cwd, Repo: x.Repo, Target: x.Target,
|
||||
Host: x.Host, Cwd: x.Cwd, Repo: x.Repo, Terminal: x.Terminal, Target: x.Target,
|
||||
Provider: x.Provider, Account: x.Account,
|
||||
Project: x.Project, Published: x.Published,
|
||||
Events: events, Children: children,
|
||||
@@ -284,6 +288,10 @@ type registerReq struct {
|
||||
Cwd string `json:"cwd"`
|
||||
Repo string `json:"repo"`
|
||||
Target string `json:"target"`
|
||||
// Terminal is the URL this session's live terminal is published at, so the
|
||||
// console can watch it. Optional — a session that publishes nothing is still
|
||||
// a session.
|
||||
Terminal string `json:"terminal"`
|
||||
// Account tag — the linked AI account this session ran under (login manager).
|
||||
Provider string `json:"provider"`
|
||||
Account string `json:"account"`
|
||||
@@ -339,6 +347,10 @@ func (o sessionOps) register(ctx context.Context, in *registerReq) (*sessionView
|
||||
if cerr != nil {
|
||||
return nil, cerr
|
||||
}
|
||||
terminal, terr := sessionTerminal(body.Terminal)
|
||||
if terr != nil {
|
||||
return nil, terr
|
||||
}
|
||||
provider := strings.TrimSpace(body.Provider)
|
||||
account := strings.TrimSpace(body.Account)
|
||||
if len(provider) > maxProvider {
|
||||
@@ -365,7 +377,7 @@ func (o sessionOps) register(ctx context.Context, in *registerReq) (*sessionView
|
||||
Title: strings.TrimSpace(body.Title),
|
||||
TaskWorkflowID: strings.TrimSpace(body.TaskWorkflowID),
|
||||
TaskRunID: strings.TrimSpace(body.TaskRunID),
|
||||
Host: host, Cwd: cwd, Repo: repo, Target: target,
|
||||
Host: host, Cwd: cwd, Repo: repo, Terminal: terminal, Target: target,
|
||||
Provider: provider, Account: account,
|
||||
Project: project, Published: body.Published,
|
||||
StartedAt: now, CreatedAt: now, UpdatedAt: now,
|
||||
@@ -407,6 +419,24 @@ func (o sessionOps) register(ctx context.Context, in *registerReq) (*sessionView
|
||||
// target). Target, when set, MUST resolve to a run-target in the SAME org (fail-
|
||||
// closed, exactly like a parent session) so a session can never claim to run on
|
||||
// another tenant's machine — the #48 dispatch association is tenant-safe.
|
||||
// sessionTerminal bounds and checks the published terminal URL. It must be https:
|
||||
// the console FRAMES this value, so anything else is a way to get a javascript:
|
||||
// or file: URL rendered on a signed-in page. Empty is fine — a session that
|
||||
// publishes no terminal simply cannot be watched.
|
||||
func sessionTerminal(v string) (string, error) {
|
||||
v = strings.TrimSpace(v)
|
||||
if v == "" {
|
||||
return "", nil
|
||||
}
|
||||
if len(v) > maxTerminal {
|
||||
return "", zip.ErrBadRequest("terminal url too long")
|
||||
}
|
||||
if !strings.HasPrefix(v, "https://") {
|
||||
return "", zip.ErrBadRequest("terminal url must be https")
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
func sessionContext(ctx context.Context, s *cloud.Service[state], org, host, cwd, repo, target string) (string, string, string, string, error) {
|
||||
host = strings.TrimSpace(host)
|
||||
if len(host) > maxHost {
|
||||
@@ -607,6 +637,10 @@ type patchSessionIn struct {
|
||||
Title *string `json:"title"`
|
||||
// Target re-dispatches a session to a run-target (the #48 association). "" detaches.
|
||||
Target *string `json:"target"`
|
||||
// Terminal publishes (or, with "", withdraws) the URL this session's live
|
||||
// terminal can be watched at. A pointer so "absent" and "withdrawn" are
|
||||
// different requests: a session that stops sharing must be able to say so.
|
||||
Terminal *string `json:"terminal"`
|
||||
// Project tags the product this session built; Published is the author's
|
||||
// decision to let anyone read the story (provenance.go). Both are pointers so
|
||||
// "absent" and "cleared" are different requests.
|
||||
@@ -688,6 +722,13 @@ func (o sessionOps) patch(ctx context.Context, in *patchSessionIn) (*sessionView
|
||||
}
|
||||
x.Target = nt // "" detaches
|
||||
}
|
||||
if body.Terminal != nil {
|
||||
nt, terr := sessionTerminal(*body.Terminal)
|
||||
if terr != nil {
|
||||
return nil, terr
|
||||
}
|
||||
x.Terminal = nt // "" withdraws
|
||||
}
|
||||
x.UpdatedAt = time.Now().Unix()
|
||||
if err := s.State.store.UpdateSession(ctx, x); err != nil {
|
||||
if err == errSessionNotFound {
|
||||
|
||||
@@ -53,6 +53,13 @@ type Session struct {
|
||||
Repo string
|
||||
Target string
|
||||
|
||||
// Terminal is where this session's live terminal can be WATCHED — the URL the
|
||||
// machine published for it (zrok gives one without opening a port). Optional:
|
||||
// a session that publishes nothing is still a session, it just cannot be
|
||||
// watched. It is a URL rather than a stream because the bytes belong to the
|
||||
// machine running the shell; cloud holds the address, never the connection.
|
||||
Terminal string
|
||||
|
||||
// Provider/Account tag a session with the linked AI account it ran under (the
|
||||
// login-manager tie-in): which provider (claude|codex|hanzo|…) and which
|
||||
// subscription/api account served this run. Optional (a surface that doesn't
|
||||
@@ -165,6 +172,7 @@ CREATE INDEX IF NOT EXISTS ix_events_org_session_seq ON agent_session_events(org
|
||||
"cwd": "TEXT NOT NULL DEFAULT ''",
|
||||
"repo": "TEXT NOT NULL DEFAULT ''",
|
||||
"target": "TEXT NOT NULL DEFAULT ''",
|
||||
"terminal": "TEXT NOT NULL DEFAULT ''",
|
||||
"provider": "TEXT NOT NULL DEFAULT ''",
|
||||
"account": "TEXT NOT NULL DEFAULT ''",
|
||||
// The readable build (provenance.go). Defaults keep every pre-existing
|
||||
@@ -189,13 +197,13 @@ CREATE INDEX IF NOT EXISTS ix_sessions_published ON agent_sessions(published, up
|
||||
return nil
|
||||
}
|
||||
|
||||
const sessionCols = `id,org,agent,actor,status,parent_id,root_id,title,started_at,ended_at,created_at,updated_at,task_workflow_id,task_run_id,host,cwd,repo,target,provider,account,project,published`
|
||||
const sessionCols = `id,org,agent,actor,status,parent_id,root_id,title,started_at,ended_at,created_at,updated_at,task_workflow_id,task_run_id,host,cwd,repo,terminal,target,provider,account,project,published`
|
||||
|
||||
func scanSession(sc interface{ Scan(...any) error }) (Session, error) {
|
||||
var x Session
|
||||
err := sc.Scan(&x.ID, &x.Org, &x.Agent, &x.Actor, &x.Status, &x.ParentID, &x.RootID,
|
||||
&x.Title, &x.StartedAt, &x.EndedAt, &x.CreatedAt, &x.UpdatedAt,
|
||||
&x.TaskWorkflowID, &x.TaskRunID, &x.Host, &x.Cwd, &x.Repo, &x.Target,
|
||||
&x.TaskWorkflowID, &x.TaskRunID, &x.Host, &x.Cwd, &x.Repo, &x.Terminal, &x.Target,
|
||||
&x.Provider, &x.Account, &x.Project, &x.Published)
|
||||
return x, err
|
||||
}
|
||||
@@ -221,10 +229,10 @@ func (s *Store) CreateSession(ctx context.Context, x Session) error {
|
||||
}
|
||||
}
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
`INSERT INTO agent_sessions (`+sessionCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||||
`INSERT INTO agent_sessions (`+sessionCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||||
x.ID, x.Org, x.Agent, x.Actor, x.Status, x.ParentID, x.RootID, x.Title,
|
||||
x.StartedAt, x.EndedAt, x.CreatedAt, x.UpdatedAt, x.TaskWorkflowID, x.TaskRunID,
|
||||
x.Host, x.Cwd, x.Repo, x.Target, x.Provider, x.Account, x.Project, x.Published)
|
||||
x.Host, x.Cwd, x.Repo, x.Terminal, x.Target, x.Provider, x.Account, x.Project, x.Published)
|
||||
if err != nil {
|
||||
return fmt.Errorf("insert session: %w", err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
package agents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A published terminal URL survives the round trip. It is execution context like
|
||||
// host and cwd, so it stores and reads back the same way — the point of the field
|
||||
// is that a console can find the terminal again later, not just at register time.
|
||||
func TestTerminalRoundTripsThroughTheStore(t *testing.T) {
|
||||
s := testSessionStore(t)
|
||||
ctx := context.Background()
|
||||
|
||||
x := mkSession("hanzo", "sess_t", "", "sess_t")
|
||||
x.Terminal = "https://abc.share.hanzo.ai"
|
||||
if err := s.CreateSession(ctx, x); err != nil {
|
||||
t.Fatalf("CreateSession: %v", err)
|
||||
}
|
||||
|
||||
got, err := s.GetSession(ctx, "hanzo", "sess_t")
|
||||
if err != nil {
|
||||
t.Fatalf("GetSession: %v", err)
|
||||
}
|
||||
if got.Terminal != x.Terminal {
|
||||
t.Fatalf("terminal = %q, want %q", got.Terminal, x.Terminal)
|
||||
}
|
||||
}
|
||||
|
||||
// The console FRAMES this value, so a scheme it would not open must be refused at
|
||||
// the edge rather than rendered on a signed-in page.
|
||||
func TestSessionTerminalRefusesNonHTTPS(t *testing.T) {
|
||||
for _, bad := range []string{
|
||||
"javascript:alert(1)",
|
||||
"file:///etc/passwd",
|
||||
"http://plain.example.com",
|
||||
"//protocol-relative.example.com",
|
||||
} {
|
||||
if _, err := sessionTerminal(bad); err == nil {
|
||||
t.Errorf("sessionTerminal(%q) was accepted; want refused", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Empty is not an error: a session that publishes no terminal is still a session,
|
||||
// it simply cannot be watched.
|
||||
func TestSessionTerminalAllowsEmpty(t *testing.T) {
|
||||
got, err := sessionTerminal(" ")
|
||||
if err != nil {
|
||||
t.Fatalf("sessionTerminal(empty): %v", err)
|
||||
}
|
||||
if got != "" {
|
||||
t.Fatalf("got %q, want empty", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionTerminalAcceptsHTTPS(t *testing.T) {
|
||||
got, err := sessionTerminal(" https://abc.share.hanzo.ai ")
|
||||
if err != nil {
|
||||
t.Fatalf("sessionTerminal(https): %v", err)
|
||||
}
|
||||
if got != "https://abc.share.hanzo.ai" {
|
||||
t.Fatalf("got %q, want the trimmed url", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A URL long enough to be a payload rather than an address is refused.
|
||||
func TestSessionTerminalIsBounded(t *testing.T) {
|
||||
long := "https://" + string(make([]byte, maxTerminal))
|
||||
if _, err := sessionTerminal(long); err == nil {
|
||||
t.Fatal("an over-long terminal url was accepted")
|
||||
}
|
||||
}
|
||||
@@ -1,234 +0,0 @@
|
||||
// Package sessions is the live-coding-session plane: the /v1/sessions surface
|
||||
// behind the session list in the console, and the answer to "what is being worked
|
||||
// on right now, and can I watch it".
|
||||
//
|
||||
// A coding session runs on a developer's own machine, not in the cluster, so the
|
||||
// cluster cannot enumerate them — the machine has to say so. A host agent starts
|
||||
// a terminal, publishes it through zrok (which is what gives it a public
|
||||
// https://<share>.share.hanzo.ai URL without opening a port), and beats here.
|
||||
// This surface holds the roster; it never proxies the terminal itself.
|
||||
//
|
||||
// Surface (org-scoped; /v1 only):
|
||||
//
|
||||
// GET /v1/sessions the caller's live sessions -> sessionsView
|
||||
// POST /v1/sessions register or heartbeat one -> Session
|
||||
// DELETE /v1/sessions/:id deregister on exit -> 204
|
||||
//
|
||||
// LIVENESS IS A TTL, NOT A STATE MACHINE. A session is live if it beat within
|
||||
// SessionTTL. There is no "stopped" transition to get wrong: a laptop that sleeps
|
||||
// mid-session stops beating and drops off the list, and reappears when it wakes.
|
||||
// Nothing has to observe the death for the roster to be right.
|
||||
//
|
||||
// ORG ISOLATION is enforced SERVER-SIDE on every request, from the validated
|
||||
// principal, and is the mandatory predicate on every store statement. It is never
|
||||
// read from a query param or body. A session URL is a live shell on someone's
|
||||
// machine; there is no cross-org read path, and no admin override.
|
||||
package sessions
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/cloud"
|
||||
"github.com/hanzoai/cloud/apps/principal"
|
||||
luxlog "github.com/luxfi/log"
|
||||
"github.com/zap-proto/zip"
|
||||
)
|
||||
|
||||
// SessionTTL is how long a session stays listed after its last heartbeat. Agents
|
||||
// beat well inside it, so one dropped beat over a flaky link does not blink a
|
||||
// session out of the UI.
|
||||
const SessionTTL = 90 * time.Second
|
||||
|
||||
// pruneAfter is when a dead row is deleted outright. Reads already ignore
|
||||
// anything past SessionTTL; this only bounds the file.
|
||||
const pruneAfter = 24 * time.Hour
|
||||
|
||||
// maxField bounds every string a host sends. A host agent is trusted to be
|
||||
// truthful about its own machine, not to be well behaved about lengths.
|
||||
const maxField = 256
|
||||
|
||||
type service struct {
|
||||
store *Store
|
||||
log luxlog.Logger
|
||||
}
|
||||
|
||||
var mounted *service
|
||||
|
||||
// sessionsView is the wire shape for a list. The TTL travels with it so a client
|
||||
// can grey out a session that is about to age out instead of hardcoding a guess.
|
||||
type sessionsView struct {
|
||||
Sessions []Session `json:"sessions"`
|
||||
TTLSeconds int `json:"ttlSeconds"`
|
||||
}
|
||||
|
||||
// Mount registers the sessions surface on app per HIP-0106.
|
||||
func Mount(app cloud.Router, deps cloud.Deps) error {
|
||||
if app == nil {
|
||||
return fmt.Errorf("sessions.Mount: nil app")
|
||||
}
|
||||
if deps.Logger == nil {
|
||||
return fmt.Errorf("sessions.Mount: nil deps.Logger")
|
||||
}
|
||||
log := deps.Logger.New("subsystem", "sessions")
|
||||
if deps.DataDir == "" {
|
||||
return fmt.Errorf("sessions.Mount: empty DataDir")
|
||||
}
|
||||
if err := os.MkdirAll(deps.DataDir, 0o755); err != nil {
|
||||
return fmt.Errorf("sessions.Mount: data dir: %w", err)
|
||||
}
|
||||
store, err := openStore(filepath.Join(deps.DataDir, "sessions.db"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("sessions.Mount: open sessions store: %w", err)
|
||||
}
|
||||
s := &service{store: store, log: log}
|
||||
mounted = s
|
||||
|
||||
g := app.Group("/v1/sessions")
|
||||
g.Get("", s.list)
|
||||
g.Post("", s.beat)
|
||||
g.Delete("/:id", s.remove)
|
||||
|
||||
log.Info("sessions surface mounted", "prefix", "/v1/sessions", "brand", deps.Brand)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Shutdown releases the sessions store. Idempotent.
|
||||
func Shutdown(_ context.Context) error {
|
||||
if mounted == nil {
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
if mounted.store != nil {
|
||||
err = mounted.store.Close()
|
||||
}
|
||||
mounted = nil
|
||||
return err
|
||||
}
|
||||
|
||||
// subject resolves the ORG that owns a session — the isolation KEY — for a
|
||||
// VALIDATED principal only. Fails closed for an unvalidated request.
|
||||
//
|
||||
// Sessions key on the ORG, not the individual, because watching a teammate's
|
||||
// build is the point of the surface. That is the one deliberate difference from
|
||||
// prefs, which keys on the person because nobody else has a reason to read a
|
||||
// theme. A user with no org yet keys on their own name, which is correct for
|
||||
// exactly as long as they have no org to be qualified by.
|
||||
func (s *service) subject(c *zip.Ctx) (string, bool) {
|
||||
if !principal.Validated(c) {
|
||||
return "", false
|
||||
}
|
||||
if owner := strings.TrimSpace(c.Org()); owner != "" && len(owner) <= principal.MaxOrgLen {
|
||||
return owner, true
|
||||
}
|
||||
name := strings.TrimSpace(c.User())
|
||||
if name == "" || len(name) > principal.MaxOrgLen {
|
||||
return "", false
|
||||
}
|
||||
return name, true
|
||||
}
|
||||
|
||||
func (s *service) list(c *zip.Ctx) error {
|
||||
subject, ok := s.subject(c)
|
||||
if !ok {
|
||||
return zip.ErrForbidden("a validated principal is required")
|
||||
}
|
||||
now := time.Now()
|
||||
out, err := s.store.List(subject, now, SessionTTL)
|
||||
if err != nil {
|
||||
s.log.Error("list sessions", "err", err)
|
||||
return zip.ErrInternal("could not list sessions")
|
||||
}
|
||||
if out == nil {
|
||||
out = []Session{}
|
||||
}
|
||||
return c.JSON(http.StatusOK, sessionsView{Sessions: out, TTLSeconds: int(SessionTTL.Seconds())})
|
||||
}
|
||||
|
||||
// beatBody is what a host agent sends. Everything except id and url is
|
||||
// descriptive: the roster is still correct without it, just less useful.
|
||||
type beatBody struct {
|
||||
ID string `json:"id"`
|
||||
Host string `json:"host"`
|
||||
Workspace string `json:"workspace"`
|
||||
Repo string `json:"repo"`
|
||||
Branch string `json:"branch"`
|
||||
Agent string `json:"agent"`
|
||||
URL string `json:"url"`
|
||||
}
|
||||
|
||||
func (s *service) beat(c *zip.Ctx) error {
|
||||
subject, ok := s.subject(c)
|
||||
if !ok {
|
||||
return zip.ErrForbidden("a validated principal is required")
|
||||
}
|
||||
var body beatBody
|
||||
if err := json.Unmarshal(c.Body(), &body); err != nil {
|
||||
return zip.ErrBadRequest("body must be a JSON object")
|
||||
}
|
||||
id := clip(body.ID)
|
||||
if id == "" {
|
||||
return zip.ErrBadRequest("id is required")
|
||||
}
|
||||
// The URL is what the console will frame, so it must be a scheme we would
|
||||
// actually open. Anything else is a way to get a javascript: or file: URL
|
||||
// rendered as a link on a signed-in page.
|
||||
url := clip(body.URL)
|
||||
if !strings.HasPrefix(url, "https://") {
|
||||
return zip.ErrBadRequest("url must be https")
|
||||
}
|
||||
|
||||
now := time.Now().Unix()
|
||||
v := Session{
|
||||
ID: id,
|
||||
Subject: subject,
|
||||
Host: clip(body.Host),
|
||||
Workspace: clip(body.Workspace),
|
||||
Repo: clip(body.Repo),
|
||||
Branch: clip(body.Branch),
|
||||
Agent: clip(body.Agent),
|
||||
URL: url,
|
||||
StartedAt: now,
|
||||
BeatAt: now,
|
||||
}
|
||||
if err := s.store.Beat(v); err != nil {
|
||||
s.log.Error("beat session", "err", err, "id", id)
|
||||
return zip.ErrInternal("could not record session")
|
||||
}
|
||||
// Opportunistic: pruning on write keeps the file bounded without a timer, and
|
||||
// a failure here has no bearing on the beat that just succeeded.
|
||||
if err := s.store.Prune(time.Now(), pruneAfter); err != nil {
|
||||
s.log.Debug("prune sessions", "err", err)
|
||||
}
|
||||
return c.JSON(http.StatusOK, v)
|
||||
}
|
||||
|
||||
func (s *service) remove(c *zip.Ctx) error {
|
||||
subject, ok := s.subject(c)
|
||||
if !ok {
|
||||
return zip.ErrForbidden("a validated principal is required")
|
||||
}
|
||||
id := clip(c.Param("id"))
|
||||
if id == "" {
|
||||
return zip.ErrBadRequest("id is required")
|
||||
}
|
||||
if err := s.store.Delete(subject, id); err != nil {
|
||||
s.log.Error("delete session", "err", err, "id", id)
|
||||
return zip.ErrInternal("could not remove session")
|
||||
}
|
||||
return c.NoContent(http.StatusNoContent)
|
||||
}
|
||||
|
||||
func clip(v string) string {
|
||||
v = strings.TrimSpace(v)
|
||||
if len(v) > maxField {
|
||||
return v[:maxField]
|
||||
}
|
||||
return v
|
||||
}
|
||||
@@ -1,141 +0,0 @@
|
||||
package sessions
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
// github.com/hanzoai/sqlite is the ONE Hanzo SQLite driver (registers the
|
||||
// "sqlite" database/sql name under both cgo and pure-Go build tags). Blank
|
||||
// import registers the driver; importing modernc directly would double-register.
|
||||
"github.com/hanzoai/cloud/cek"
|
||||
_ "github.com/hanzoai/sqlite"
|
||||
)
|
||||
|
||||
// The sessions STORE holds one row per live coding session, following the
|
||||
// settings-store discipline — Hanzo Base/SQLite, MaxOpenConns(1) to serialize
|
||||
// writes against the file lock, one file.
|
||||
//
|
||||
// Isolation is `subject` on every statement: a session is visible to the org that
|
||||
// registered it and to nobody else. A terminal is a live shell on a developer's
|
||||
// machine, so cross-tenant reads are not a feature to be added later.
|
||||
type Store struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
// Session is one live coding session. ID is chosen by the host agent and is
|
||||
// stable across heartbeats, so re-registering the same session updates a row
|
||||
// instead of accumulating duplicates.
|
||||
type Session struct {
|
||||
ID string `json:"id"`
|
||||
Subject string `json:"-"`
|
||||
Host string `json:"host"`
|
||||
Workspace string `json:"workspace"`
|
||||
Repo string `json:"repo,omitempty"`
|
||||
Branch string `json:"branch,omitempty"`
|
||||
Agent string `json:"agent,omitempty"`
|
||||
URL string `json:"url"`
|
||||
StartedAt int64 `json:"startedAt"`
|
||||
BeatAt int64 `json:"beatAt"`
|
||||
}
|
||||
|
||||
func openStore(path string) (*Store, error) {
|
||||
db, err := cek.Open(cek.Global, path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open sqlite %q: %w", path, err)
|
||||
}
|
||||
db.SetMaxOpenConns(1)
|
||||
for _, pragma := range []string{
|
||||
"PRAGMA busy_timeout=5000",
|
||||
"PRAGMA journal_mode=WAL",
|
||||
"PRAGMA foreign_keys=ON",
|
||||
} {
|
||||
if _, err := db.Exec(pragma); err != nil {
|
||||
_ = db.Close()
|
||||
return nil, fmt.Errorf("pragma %q: %w", pragma, err)
|
||||
}
|
||||
}
|
||||
s := &Store{db: db}
|
||||
if err := s.migrate(); err != nil {
|
||||
_ = db.Close()
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Store) migrate() error {
|
||||
const ddl = `
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
subject TEXT NOT NULL,
|
||||
id TEXT NOT NULL,
|
||||
host TEXT NOT NULL,
|
||||
workspace TEXT NOT NULL,
|
||||
repo TEXT NOT NULL DEFAULT '',
|
||||
branch TEXT NOT NULL DEFAULT '',
|
||||
agent TEXT NOT NULL DEFAULT '',
|
||||
url TEXT NOT NULL,
|
||||
started_at INTEGER NOT NULL,
|
||||
beat_at INTEGER NOT NULL,
|
||||
PRIMARY KEY (subject, id)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS sessions_beat ON sessions(beat_at);`
|
||||
_, err := s.db.Exec(ddl)
|
||||
return err
|
||||
}
|
||||
|
||||
// Beat upserts a session. started_at survives a heartbeat so the UI can show how
|
||||
// long a session has been running; every other field is refreshed, because a
|
||||
// session that moves branch mid-run should say so.
|
||||
func (s *Store) Beat(v Session) error {
|
||||
const q = `
|
||||
INSERT INTO sessions (subject, id, host, workspace, repo, branch, agent, url, started_at, beat_at)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT(subject, id) DO UPDATE SET
|
||||
host=excluded.host, workspace=excluded.workspace, repo=excluded.repo,
|
||||
branch=excluded.branch, agent=excluded.agent, url=excluded.url,
|
||||
beat_at=excluded.beat_at`
|
||||
_, err := s.db.Exec(q, v.Subject, v.ID, v.Host, v.Workspace, v.Repo, v.Branch,
|
||||
v.Agent, v.URL, v.StartedAt, v.BeatAt)
|
||||
return err
|
||||
}
|
||||
|
||||
// List returns the caller's sessions that have beaten within ttl, newest beat
|
||||
// first. Liveness is a read-time predicate rather than a reaper: a host that
|
||||
// loses power stops beating, and stops being listed, without anything having to
|
||||
// notice it died.
|
||||
func (s *Store) List(subject string, now time.Time, ttl time.Duration) ([]Session, error) {
|
||||
const q = `
|
||||
SELECT id, host, workspace, repo, branch, agent, url, started_at, beat_at
|
||||
FROM sessions WHERE subject=? AND beat_at >= ? ORDER BY beat_at DESC`
|
||||
rows, err := s.db.Query(q, subject, now.Add(-ttl).Unix())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Session
|
||||
for rows.Next() {
|
||||
v := Session{Subject: subject}
|
||||
if err := rows.Scan(&v.ID, &v.Host, &v.Workspace, &v.Repo, &v.Branch,
|
||||
&v.Agent, &v.URL, &v.StartedAt, &v.BeatAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, v)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// Delete removes one of the caller's sessions. Deleting a session that is not
|
||||
// there is not an error: a host that exits twice should not have to care.
|
||||
func (s *Store) Delete(subject, id string) error {
|
||||
_, err := s.db.Exec(`DELETE FROM sessions WHERE subject=? AND id=?`, subject, id)
|
||||
return err
|
||||
}
|
||||
|
||||
// Prune drops rows that stopped beating long ago. Reads already ignore them; this
|
||||
// only keeps the file from growing without bound.
|
||||
func (s *Store) Prune(now time.Time, keep time.Duration) error {
|
||||
_, err := s.db.Exec(`DELETE FROM sessions WHERE beat_at < ?`, now.Add(-keep).Unix())
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Store) Close() error { return s.db.Close() }
|
||||
@@ -1,141 +0,0 @@
|
||||
package sessions
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/cloud/cek"
|
||||
)
|
||||
|
||||
func testStore(t *testing.T) *Store {
|
||||
t.Helper()
|
||||
// the data plane refuses to open unencrypted, so a test supplies its own key
|
||||
cek.SetMasterKey(bytes.Repeat([]byte{0x2a}, 32))
|
||||
s, err := openStore(filepath.Join(t.TempDir(), "sessions.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("openStore: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = s.Close() })
|
||||
return s
|
||||
}
|
||||
|
||||
func beat(t *testing.T, s *Store, subject, id string, at time.Time) {
|
||||
t.Helper()
|
||||
if err := s.Beat(Session{
|
||||
Subject: subject, ID: id, Host: "dbc", Workspace: "/w",
|
||||
URL: "https://x.share.hanzo.ai", StartedAt: at.Unix(), BeatAt: at.Unix(),
|
||||
}); err != nil {
|
||||
t.Fatalf("Beat(%s/%s): %v", subject, id, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A session that beats twice is still one session. Without the upsert, every
|
||||
// heartbeat would add a row and the console would show one terminal N times.
|
||||
func TestBeatIsUpsertNotInsert(t *testing.T) {
|
||||
s := testStore(t)
|
||||
now := time.Now()
|
||||
beat(t, s, "hanzo", "a", now)
|
||||
beat(t, s, "hanzo", "a", now.Add(time.Second))
|
||||
|
||||
got, err := s.List("hanzo", now.Add(time.Second), SessionTTL)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("got %d sessions after two beats, want 1", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
// Liveness is a read-time predicate: a host that stops beating drops off the
|
||||
// roster on its own, with nothing having to observe that it died.
|
||||
func TestListHidesSessionsPastTTL(t *testing.T) {
|
||||
s := testStore(t)
|
||||
start := time.Now()
|
||||
beat(t, s, "hanzo", "stale", start)
|
||||
beat(t, s, "hanzo", "fresh", start.Add(SessionTTL))
|
||||
|
||||
// read one second past the fresh beat, so the cutoff lands after the stale
|
||||
// one. A session exactly on the cutoff counts as live — the boundary is
|
||||
// inclusive, and a beat that arrives right on the TTL is not a dead host.
|
||||
got, err := s.List("hanzo", start.Add(SessionTTL+time.Second), SessionTTL)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
if len(got) != 1 || got[0].ID != "fresh" {
|
||||
t.Fatalf("got %+v, want only the session that beat within the TTL", ids(got))
|
||||
}
|
||||
}
|
||||
|
||||
// A session URL is a live shell on someone's machine. One org must never see
|
||||
// another's, and the predicate belongs in the query, not the caller.
|
||||
func TestListIsScopedToSubject(t *testing.T) {
|
||||
s := testStore(t)
|
||||
now := time.Now()
|
||||
beat(t, s, "hanzo", "mine", now)
|
||||
beat(t, s, "zoo", "theirs", now)
|
||||
|
||||
got, err := s.List("hanzo", now, SessionTTL)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
if len(got) != 1 || got[0].ID != "mine" {
|
||||
t.Fatalf("got %v, want only the caller's own session", ids(got))
|
||||
}
|
||||
}
|
||||
|
||||
// Delete is likewise scoped: knowing another org's session id must not be enough
|
||||
// to remove it.
|
||||
func TestDeleteCannotReachAnotherSubject(t *testing.T) {
|
||||
s := testStore(t)
|
||||
now := time.Now()
|
||||
beat(t, s, "zoo", "theirs", now)
|
||||
|
||||
if err := s.Delete("hanzo", "theirs"); err != nil {
|
||||
t.Fatalf("Delete: %v", err)
|
||||
}
|
||||
got, err := s.List("zoo", now, SessionTTL)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("another subject's delete removed the row")
|
||||
}
|
||||
}
|
||||
|
||||
// Deleting a session that is not there is not an error: a host that exits twice
|
||||
// should not have to care.
|
||||
func TestDeleteMissingIsNotAnError(t *testing.T) {
|
||||
if err := testStore(t).Delete("hanzo", "nope"); err != nil {
|
||||
t.Fatalf("Delete(missing): %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPruneDropsLongDeadRows(t *testing.T) {
|
||||
s := testStore(t)
|
||||
start := time.Now()
|
||||
beat(t, s, "hanzo", "ancient", start)
|
||||
now := start.Add(pruneAfter + time.Minute)
|
||||
|
||||
if err := s.Prune(now, pruneAfter); err != nil {
|
||||
t.Fatalf("Prune: %v", err)
|
||||
}
|
||||
// look with a TTL wide enough to have found it, so the assertion is about
|
||||
// pruning rather than about the liveness window
|
||||
got, err := s.List("hanzo", now, pruneAfter*2)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
if len(got) != 0 {
|
||||
t.Fatalf("got %v, want the long-dead row pruned", ids(got))
|
||||
}
|
||||
}
|
||||
|
||||
func ids(v []Session) []string {
|
||||
out := make([]string, 0, len(v))
|
||||
for _, s := range v {
|
||||
out = append(out, s.ID)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -137,7 +137,6 @@ var Apps = []App{
|
||||
{Name: "meet", Prefixes: []string{"/v1/meet/getToken", "/v1/meet/health"}},
|
||||
{Name: "settings", Prefixes: []string{"/v1/settings"}},
|
||||
{Name: "prefs", Prefixes: []string{"/v1/prefs"}},
|
||||
{Name: "sessions", Prefixes: []string{"/v1/sessions"}},
|
||||
{Name: "notify", Prefixes: []string{"/v1/notify"}},
|
||||
{Name: "channels", Prefixes: []string{"/v1/channels"}},
|
||||
{Name: "gateway", Prefixes: []string{"/v1/gateway"}},
|
||||
|
||||
@@ -27,7 +27,7 @@ var frozen = []string{
|
||||
"campaign", "validators", "social", "analytics", "git", "sync",
|
||||
"visor", "venue", "captable", "code", "zero-trust", "share",
|
||||
"dataroom", "graph", "security", "integrations", "destinations", "cloudflare",
|
||||
"sbom", "team", "meet", "settings", "prefs", "sessions", "notify",
|
||||
"sbom", "team", "meet", "settings", "prefs", "notify",
|
||||
"channels", "gateway", "entitlements", "exec", "websearch", "crawl",
|
||||
"index", "catalog", "world", "bot", "runtime", "authors",
|
||||
"bots", "audit", "affiliates", "esign", "product", "evals",
|
||||
|
||||
@@ -1,28 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"github.com/hanzoai/cloud"
|
||||
"github.com/hanzoai/cloud/apps/sessions"
|
||||
)
|
||||
|
||||
// Standalone entry for the sessions app.
|
||||
//
|
||||
// This is the app's OWN composition root: it links only its own subsystem and
|
||||
// the cloud request tier, never the whole fleet, so the build is this one app
|
||||
// and not the ~3040-package union the fused binary was. The light host loads it
|
||||
// as a plugin; run directly it serves standalone. Its OpenAPI subset comes from
|
||||
// `sessions openapi`. Hand-owned — edit the spec below directly.
|
||||
func main() {
|
||||
if err := cloud.Serve([]cloud.Plugin{{
|
||||
Name: "sessions",
|
||||
Price: cloud.Free,
|
||||
Mount: sessions.Mount,
|
||||
Shutdown: sessions.Shutdown,
|
||||
}}, []string{"sessions"}); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
@@ -1,10 +0,0 @@
|
||||
[
|
||||
{
|
||||
"description": "ListSessions returns the coding sessions on the caller's org machines that are live right now \u2014 host, workspace, repo and branch, plus the URL of the terminal each one is publishing. Liveness is a heartbeat TTL, so a machine that sleeps or loses its link drops off this list and returns when it comes back.",
|
||||
"inputSchema": {
|
||||
"properties": {},
|
||||
"type": "object"
|
||||
},
|
||||
"name": "get_v1_sessions"
|
||||
}
|
||||
]
|
||||
@@ -1,42 +0,0 @@
|
||||
{
|
||||
"openapi": "3.1.0",
|
||||
"info": {
|
||||
"title": "Hanzo Cloud API",
|
||||
"description": "Generated from the live router \u2014 every operation below is a route the unified cloud binary actually serves. Tagged by product: the first path segment after /v1/.",
|
||||
"version": "v1"
|
||||
},
|
||||
"servers": [
|
||||
{
|
||||
"url": "https://api.hanzo.ai"
|
||||
}
|
||||
],
|
||||
"tags": [
|
||||
{
|
||||
"name": "sessions"
|
||||
}
|
||||
],
|
||||
"paths": {
|
||||
"/v1/sessions": {
|
||||
"get": {
|
||||
"operationId": "get_v1_sessions",
|
||||
"tags": [
|
||||
"sessions"
|
||||
]
|
||||
},
|
||||
"post": {
|
||||
"operationId": "post_v1_sessions",
|
||||
"tags": [
|
||||
"sessions"
|
||||
]
|
||||
}
|
||||
},
|
||||
"/v1/sessions/:id": {
|
||||
"delete": {
|
||||
"operationId": "delete_v1_sessions_id",
|
||||
"tags": [
|
||||
"sessions"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user