From c79104bd103a7b8a37ae6242fad1e03bd990e223 Mon Sep 17 00:00:00 2001 From: Michael Pursifull Date: Wed, 5 Aug 2026 13:47:21 -0500 Subject: [PATCH] feat(session): cooperative context feed for interactive claude via statusline hooks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Interactive (TUI) sessions emit no parseable stream, so the usage accountant cannot meter them and CTX% renders "-". This closes the gap using the harness's own statusline side-channel (finding-011): - runtime.context_feed = "statusline" (manifest, validated) opts a role into the feed - the projection layer merges marvel-owned statusLine and subagentStatusLine hooks into the projected settings; policy content is never modified and policy keys always win - the injected statusLine carries refreshInterval 15 so an idle session keeps beating instead of starving a heartbeat healthcheck - new hidden command `marvel ctx-forward` reads the statusline payload on stdin, prints a compact status for the human attached to the pane, and forwards used_percentage to the heartbeat RPC using the MARVEL_SOCKET/WORKSPACE/SESSION env the adapter already injects; subagent payloads render a summary and send nothing (no daemon surface for per-subagent context yet) Live-verified end to end: interactive claude session under examples/context-feed.toml shows CTX% 13% in `marvel get sessions` after one model turn, with the pane statusline reading "Fable 5 · CTX 13% · $1.47". Refs: aae-orc-7hzb, aae-orc-dc1j --- CLAUDE.md | 2 +- cmd/marvel/ctxforward.go | 130 +++++++++++++++++++++++++ cmd/marvel/ctxforward_test.go | 70 ++++++++++++++ cmd/marvel/main.go | 1 + examples/context-feed.toml | 41 ++++++++ internal/api/manifest.go | 9 ++ internal/api/types.go | 11 +++ internal/session/projection.go | 69 +++++++++++--- internal/session/projection_test.go | 142 ++++++++++++++++++++++++++++ 9 files changed, 463 insertions(+), 12 deletions(-) create mode 100644 cmd/marvel/ctxforward.go create mode 100644 cmd/marvel/ctxforward_test.go create mode 100644 examples/context-feed.toml diff --git a/CLAUDE.md b/CLAUDE.md index 977bbce..1e2f49b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -58,7 +58,7 @@ Declared in TOML or YAML manifests, applied with `marvel work`. |---|---|---| | Namespace | **Workspace** | Isolation boundary: a project, team, or environment. Scopes every other resource. | | Pod | **Session** | The atomic unit. One tmux pane running one harness process. Lifecycle pending → running → succeeded/failed, plus crashed and crashloop-backoff. Restartable. | -| Container | **Runtime** | The harness binary, args, and mode, plus `context_window` to override the model-to-window table for CTX%. `runtime` names the HARNESS (claude, codex, opencode), never the agent: `elem-runtime-names-harness`. | +| Container | **Runtime** | The harness binary, args, and mode, plus `context_window` to override the model-to-window table for CTX%, and `context_feed = "statusline"` to give interactive claude sessions a cooperative CTX% feed via projected statusline hooks + `marvel ctx-forward` (finding-011, `examples/context-feed.toml`). `runtime` names the HARNESS (claude, codex, opencode), never the agent: `elem-runtime-names-harness`. | | Deployment | **Team** | Heterogeneous roles, each with its own runtime and replica count. Per-role scaling, shifts. Binds a supervisor to its agents. | | (none) | **Role** | One kind of agent within a team: name, replicas, runtime, restart policy, `max_restarts`, `permissions`, `dangerous_permissions`, `policy`, persona, identity. | | Service | **Endpoint** | A named record of `{name, workspace, team}` and nothing else. Created from a manifest `[[endpoint]]` section, read with `marvel get endpoints` and `marvel describe endpoint`. No role field, and no code resolves an endpoint to a session, so it is a name in the store rather than a routing target. Role-based routing waits on director (roadmap M2). | diff --git a/cmd/marvel/ctxforward.go b/cmd/marvel/ctxforward.go new file mode 100644 index 0000000..fb0eb9d --- /dev/null +++ b/cmd/marvel/ctxforward.go @@ -0,0 +1,130 @@ +package main + +import ( + "encoding/json" + "fmt" + "io" + "os" + + "github.com/arcavenae/marvel/internal/daemon" + "github.com/spf13/cobra" +) + +// ctx-forward is the statusline side of the cooperative context feed +// (finding-011). Claude Code invokes the configured statusline command +// with a JSON payload on stdin; the projection layer points that hook at +// this subcommand. It forwards the harness's own context figure to the +// daemon's heartbeat RPC and prints a compact status string for the +// human attached to the pane. +// +// Two payload shapes arrive here, distinguished by their keys: +// - the main statusLine payload (context_window object, cost object) +// - the subagentStatusLine payload (tasks array) +// +// Failure posture: this runs on every statusline tick inside an agent's +// pane, so it never exits nonzero and never prints errors — a broken +// feed shows as a silent gap in CTX%, diagnosed via `marvel describe +// session`, not as red text inside the agent's terminal. + +// statuslinePayload is the subset of Claude Code's statusline JSON that +// the forwarder reads. Fields the forwarder does not use are omitted — +// the harness owns this schema, marvel just picks out context and cost. +type statuslinePayload struct { + Model struct { + DisplayName string `json:"display_name"` + } `json:"model"` + Cost struct { + TotalCostUSD float64 `json:"total_cost_usd"` + } `json:"cost"` + ContextWindow *struct { + UsedPercentage *float64 `json:"used_percentage"` + } `json:"context_window"` + Tasks []struct { + Status string `json:"status"` + TokenCount int `json:"tokenCount"` + ContextWindowSize int `json:"contextWindowSize"` + } `json:"tasks"` +} + +// renderForward parses one statusline payload and returns the status text +// to print plus the context percentage to forward (send=false when the +// payload carries no forwardable figure). Pure so it is table-testable. +func renderForward(raw []byte) (line string, pct float64, send bool) { + var p statuslinePayload + if err := json.Unmarshal(raw, &p); err != nil { + return "marvel ctx-forward: unreadable payload", 0, false + } + + // Subagent shape: summarize the task rows. No RPC — the daemon has + // no per-subagent surface yet (tracked in aae-orc-7hzb). + if len(p.Tasks) > 0 { + running := 0 + maxPct := 0.0 + for _, t := range p.Tasks { + if t.Status == "running" { + running++ + } + if t.ContextWindowSize > 0 { + if pc := float64(t.TokenCount) / float64(t.ContextWindowSize) * 100; pc > maxPct { + maxPct = pc + } + } + } + return fmt.Sprintf("agents %d/%d running · max CTX %.0f%%", running, len(p.Tasks), maxPct), 0, false + } + + if p.ContextWindow == nil || p.ContextWindow.UsedPercentage == nil { + // Session too young to have a measurement. Show something + // stable rather than flickering an error. + return fmt.Sprintf("%s · CTX –", orUnknown(p.Model.DisplayName)), 0, false + } + pct = *p.ContextWindow.UsedPercentage + line = fmt.Sprintf("%s · CTX %.0f%% · $%.2f", orUnknown(p.Model.DisplayName), pct, p.Cost.TotalCostUSD) + return line, pct, true +} + +func orUnknown(s string) string { + if s == "" { + return "agent" + } + return s +} + +func newCtxForwardCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "ctx-forward", + Short: "Forward a statusline payload's context figure to the daemon (internal)", + Hidden: true, + RunE: func(cmd *cobra.Command, args []string) error { + raw, err := readAllStdin() + if err != nil { + fmt.Println("marvel ctx-forward") + return nil + } + line, pct, send := renderForward(raw) + fmt.Println(line) + + socket := os.Getenv("MARVEL_SOCKET") + workspace := os.Getenv("MARVEL_WORKSPACE") + session := os.Getenv("MARVEL_SESSION") + if !send || socket == "" || workspace == "" || session == "" { + return nil + } + params, _ := json.Marshal(map[string]any{ + "session_key": workspace + "/" + session, + "context_percent": pct, + }) + // Best-effort by design; see the failure posture above. + _, _ = daemon.SendRequest(socket, daemon.Request{ + Method: "heartbeat", + Params: params, + }) + return nil + }, + } + return cmd +} + +func readAllStdin() ([]byte, error) { + return io.ReadAll(os.Stdin) +} diff --git a/cmd/marvel/ctxforward_test.go b/cmd/marvel/ctxforward_test.go new file mode 100644 index 0000000..22f906d --- /dev/null +++ b/cmd/marvel/ctxforward_test.go @@ -0,0 +1,70 @@ +package main + +import ( + "strings" + "testing" +) + +func TestRenderForward(t *testing.T) { + t.Parallel() + tests := []struct { + name string + payload string + wantSend bool + wantPct float64 + wantIn string + }{ + { + name: "main payload with measurement", + payload: `{"model":{"display_name":"Haiku 4.5"}, + "cost":{"total_cost_usd":0.0898}, + "context_window":{"used_percentage":17}}`, + wantSend: true, + wantPct: 17, + wantIn: "CTX 17%", + }, + { + name: "young session, null percentage", + payload: `{"model":{"display_name":"Haiku 4.5"}, + "cost":{"total_cost_usd":0}, + "context_window":{"used_percentage":null}}`, + wantSend: false, + wantIn: "CTX –", + }, + { + name: "no context_window at all", + payload: `{"model":{"display_name":"Haiku 4.5"}}`, + wantSend: false, + wantIn: "CTX –", + }, + { + name: "subagent payload never sends", + payload: `{"tasks":[ + {"status":"running","tokenCount":11238,"contextWindowSize":200000}, + {"status":"completed","tokenCount":10818,"contextWindowSize":200000}]}`, + wantSend: false, + wantIn: "agents 1/2 running", + }, + { + name: "garbage payload", + payload: `not json`, + wantSend: false, + wantIn: "unreadable", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + line, pct, send := renderForward([]byte(tt.payload)) + if send != tt.wantSend { + t.Fatalf("send = %v, want %v", send, tt.wantSend) + } + if send && pct != tt.wantPct { + t.Errorf("pct = %v, want %v", pct, tt.wantPct) + } + if !strings.Contains(line, tt.wantIn) { + t.Errorf("line = %q, want it to contain %q", line, tt.wantIn) + } + }) + } +} diff --git a/cmd/marvel/main.go b/cmd/marvel/main.go index 41afa66..fef90c9 100644 --- a/cmd/marvel/main.go +++ b/cmd/marvel/main.go @@ -115,6 +115,7 @@ func main() { root.AddCommand(configCmd()) root.AddCommand(stopCmd()) root.AddCommand(eventsCmd()) + root.AddCommand(newCtxForwardCmd()) if err := root.Execute(); err != nil { os.Exit(1) diff --git a/examples/context-feed.toml b/examples/context-feed.toml new file mode 100644 index 0000000..a060615 --- /dev/null +++ b/examples/context-feed.toml @@ -0,0 +1,41 @@ +# Marvel example — cooperative context feed for interactive Claude Code. +# +# Interactive (TUI) sessions emit no parseable stream, so the usage +# accountant cannot meter them and CTX% renders "-". context_feed = +# "statusline" closes that gap: the projection layer injects Claude +# Code's statusLine/subagentStatusLine hooks pointing at +# `marvel ctx-forward`, which forwards the harness's OWN context figure +# (raw occupancy, with the harness's own window as denominator) to the +# heartbeat RPC. See marvel finding-011. +# +# The injected statusLine carries refreshInterval = 15 so the feed keeps +# beating while the session idles; without it, statusline updates are +# event-driven and an idle session would starve a heartbeat healthcheck. +# +# Composition rule: if the role's policy defines statusLine or +# subagentStatusLine itself, the policy wins and marvel adds nothing. +# +# Setup: +# just build +# ./bin/marvel daemon & +# ./bin/marvel work examples/context-feed.toml +# +# Verify (after the first model turn): +# ./bin/marvel get sessions # CTX% populates for the watcher +# ./bin/marvel describe session feed/watch-watcher-g1-0 + +[workspace] +name = "feed" + +[[team]] +name = "watch" + + [[team.role]] + name = "watcher" + replicas = 1 + permissions = "plan" + + [team.role.runtime] + image = "claude" + command = "claude" + context_feed = "statusline" diff --git a/internal/api/manifest.go b/internal/api/manifest.go index 2dbb9a1..b981019 100644 --- a/internal/api/manifest.go +++ b/internal/api/manifest.go @@ -143,6 +143,9 @@ type ManifestRuntime struct { Prompt string `toml:"prompt,omitempty" yaml:"prompt,omitempty"` // ContextWindow overrides the model-to-limit table, in tokens. ContextWindow int `toml:"context_window,omitempty" yaml:"context_window,omitempty"` + // ContextFeed opts an interactive session into cooperative context + // reporting. Only "statusline" is understood. See api.Runtime. + ContextFeed string `toml:"context_feed,omitempty" yaml:"context_feed,omitempty"` } // ManifestEndpoint is an endpoint section of a manifest. @@ -274,6 +277,11 @@ func validateManifest(m *Manifest) (*Manifest, error) { if r.Runtime.ContextWindow < 0 { return nil, fmt.Errorf("parse manifest: team[%d].role[%d].runtime.context_window must be >= 0", i, j) } + // Like permissions: empty means unset, but a non-empty typo + // would silently project nothing, so reject it here. + if r.Runtime.ContextFeed != "" && r.Runtime.ContextFeed != ContextFeedStatusline { + return nil, fmt.Errorf("parse manifest: team[%d].role[%d].runtime.context_feed %q is not valid (valid: %q)", i, j, r.Runtime.ContextFeed, ContextFeedStatusline) + } if r.Policy != "" && !policyNames[r.Policy] { return nil, fmt.Errorf("parse manifest: team[%d].role[%d] references undefined policy %q", i, j, r.Policy) } @@ -524,6 +532,7 @@ func (m *Manifest) Apply(store *Store) error { Mode: mr.Runtime.Mode, Prompt: mr.Runtime.Prompt, ContextWindow: mr.Runtime.ContextWindow, + ContextFeed: mr.Runtime.ContextFeed, } if rt.Name == "" { rt.Name = rt.Command diff --git a/internal/api/types.go b/internal/api/types.go index 35801ef..2bf0b6a 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -119,8 +119,19 @@ type Runtime struct { // window the harness declares itself outranks this, because the // harness's own belief is what enforces compaction. ContextWindow int `toml:"context_window,omitempty"` + // ContextFeed opts an interactive session into a cooperative context + // pressure feed. The only value today is "statusline": the projection + // layer injects statusLine/subagentStatusLine hooks pointing at + // `marvel ctx-forward`, which forwards the harness's own context + // figures to the heartbeat RPC. Headless sessions do not need this — + // their stream already feeds the usage accountant. See finding-011. + ContextFeed string `toml:"context_feed,omitempty"` } +// ContextFeedStatusline is the only ContextFeed value marvel understands +// today: statusline-hook forwarding for interactive claude sessions. +const ContextFeedStatusline = "statusline" + // Session is the atomic unit: a tmux pane running one process (pod equivalent). // // PID is tmux's pane_pid: the shell tmux started, not the agent binary diff --git a/internal/session/projection.go b/internal/session/projection.go index 112cae9..98d4af8 100644 --- a/internal/session/projection.go +++ b/internal/session/projection.go @@ -111,29 +111,46 @@ func (m *Manager) reprojectContext(sess api.Session) (*runtime.LaunchContext, ru // a file was written, and whether the written content differed from what // was already on disk. // -// A role with no policy yields wrote=false, no error. An adapter with no -// settings surface logs the policy as advisory and yields wrote=false. A -// referenced policy that is not in the store is an error (validation should -// prevent it, so reaching here means drift worth surfacing). +// A role with no policy and no context feed yields wrote=false, no error. +// An adapter with no settings surface logs the policy as advisory and +// yields wrote=false. A referenced policy that is not in the store is an +// error (validation should prevent it, so reaching here means drift worth +// surfacing). +// +// Policy content is still written unmodified — marvel never edits a key +// the policy declares. When the runtime opts into context_feed = +// "statusline", marvel ADDS its own statusLine/subagentStatusLine keys, +// and only where the policy does not define them (policy wins). See +// finding-011. func (m *Manager) projectPolicy(lctx *runtime.LaunchContext, adapter runtime.Adapter) (runtime.ProjectionTarget, bool, bool, error) { - if m.ProjectionDir == "" || lctx.Role.Policy == "" { + feed := lctx.Session.Runtime.ContextFeed == api.ContextFeedStatusline + if m.ProjectionDir == "" || (lctx.Role.Policy == "" && !feed) { return runtime.ProjectionTarget{}, false, false, nil } target := adapter.ProjectionFor(lctx, m.ProjectionDir) if !target.Supported { - log.Printf("session %s: runtime %q has no settings surface; policy %q is advisory, not projected", + log.Printf("session %s: runtime %q has no settings surface; policy %q / context feed are advisory, not projected", lctx.Session.Key(), adapter.Name(), lctx.Role.Policy) return target, false, false, nil } - key := fmt.Sprintf("%s/%s", lctx.Workspace.Name, lctx.Role.Policy) - policy, err := m.store.GetPolicy(key) - if err != nil { - return target, false, false, fmt.Errorf("resolve policy %s: %w", key, err) + settings := map[string]any{} + if lctx.Role.Policy != "" { + key := fmt.Sprintf("%s/%s", lctx.Workspace.Name, lctx.Role.Policy) + policy, err := m.store.GetPolicy(key) + if err != nil { + return target, false, false, fmt.Errorf("resolve policy %s: %w", key, err) + } + for k, v := range policy.Settings { + settings[k] = v + } + } + if feed { + injectStatuslineFeed(settings) } - changed, err := writeProjectionFile(target.Path, policy.Settings) + changed, err := writeProjectionFile(target.Path, settings) if err != nil { return target, false, false, err } @@ -145,6 +162,36 @@ func (m *Manager) projectPolicy(lctx *runtime.LaunchContext, adapter runtime.Ada // differs from what was already there, so callers can emit an event only on // a real contract change. Files are 0600: a settings fragment can carry an // allow/deny list an operator would not want world-readable. +// injectStatuslineFeed adds the statusLine/subagentStatusLine hooks that +// forward the harness's own context figures to the heartbeat RPC, keyed to +// this daemon's binary so the pane needs no PATH assumption. Policy wins: +// a key the settings document already carries is left untouched. +// +// refreshInterval keeps the feed beating while the session idles — +// statusline updates are event-driven and go quiet between prompts, which +// would otherwise starve a heartbeat healthcheck watching this session. +func injectStatuslineFeed(settings map[string]any) { + exe, err := os.Executable() + if err != nil { + log.Printf("context feed: cannot resolve marvel binary path, feed not injected: %v", err) + return + } + hook := map[string]any{ + "type": "command", + "command": exe + " ctx-forward", + "refreshInterval": 15, + } + if _, ok := settings["statusLine"]; !ok { + settings["statusLine"] = hook + } + if _, ok := settings["subagentStatusLine"]; !ok { + settings["subagentStatusLine"] = map[string]any{ + "type": "command", + "command": exe + " ctx-forward", + } + } +} + func writeProjectionFile(path string, settings map[string]any) (bool, error) { // A nil settings map projects an empty object rather than the JSON // literal null, so the harness always reads a well-formed settings file. diff --git a/internal/session/projection_test.go b/internal/session/projection_test.go index 26ae028..2ba6760 100644 --- a/internal/session/projection_test.go +++ b/internal/session/projection_test.go @@ -206,3 +206,145 @@ func TestReprojectNoPolicyIsNoOp(t *testing.T) { t.Fatalf("Reproject changed = %d, want 0 with no policy reference", n) } } + +const feedOnlyManifest = ` +[workspace] +name = "acme" + +[[team]] +name = "squad" + + [[team.role]] + name = "watcher" + replicas = 1 + + [team.role.runtime] + image = "claude" + command = "claude" + context_feed = "statusline" +` + +const feedWithPolicyManifest = ` +[workspace] +name = "acme" + +[[policy]] +name = "own-statusline" +version = "1.0" + + [policy.settings.statusLine] + type = "command" + command = "/usr/local/bin/my-statusline" + +[[team]] +name = "squad" + + [[team.role]] + name = "watcher" + replicas = 1 + policy = "own-statusline" + + [team.role.runtime] + image = "claude" + command = "claude" + context_feed = "statusline" +` + +// seedFeedSession is seedRunningSession with ContextFeed set on the +// session's runtime, the way reconcileRole copies it from the role. +func seedFeedSession(t *testing.T, mgr *Manager, workspace, team, role, name string) string { + t.Helper() + sess := &api.Session{ + Name: name, + Workspace: workspace, + Team: team, + Role: role, + Runtime: api.Runtime{Name: "claude", Command: "claude", ContextFeed: api.ContextFeedStatusline}, + } + if err := mgr.store.CreateSession(sess); err != nil { + t.Fatalf("create session: %v", err) + } + if err := mgr.store.UpdateSession(sess.Key(), func(live *api.Session) error { + live.State = api.SessionRunning + live.PaneID = "%1" + return nil + }); err != nil { + t.Fatalf("mark running: %v", err) + } + return sess.Key() +} + +// TestProjectionInjectsStatuslineFeed covers finding-011: a role with +// context_feed = "statusline" and NO policy still gets a projected +// settings file carrying the ctx-forward hooks. Falsification: with the +// old policy-only gate in projectPolicy, no file is written at all. +func TestProjectionInjectsStatuslineFeed(t *testing.T) { + t.Parallel() + mgr, _ := projectionManager(t) + + m, err := api.ParseManifestBytes([]byte(feedOnlyManifest)) + if err != nil { + t.Fatalf("parse: %v", err) + } + if err := m.Apply(mgr.store); err != nil { + t.Fatalf("apply: %v", err) + } + key := seedFeedSession(t, mgr, "acme", "squad", "watcher", "squad-watcher-g1-0") + + if n := mgr.Reproject(); n != 1 { + t.Fatalf("Reproject changed = %d, want 1", n) + } + path := filepath.Join(mgr.ProjectionDir, strings.ReplaceAll(key, "/", "-")+".settings.json") + got := readProjection(t, path) + + sl, ok := got["statusLine"].(map[string]any) + if !ok { + t.Fatalf("projection missing statusLine: %v", got) + } + cmd, _ := sl["command"].(string) + if !strings.HasSuffix(cmd, " ctx-forward") { + t.Errorf("statusLine.command = %q, want ctx-forward suffix", cmd) + } + if ri, ok := sl["refreshInterval"].(float64); !ok || ri != 15 { + t.Errorf("statusLine.refreshInterval = %v, want 15", sl["refreshInterval"]) + } + sub, ok := got["subagentStatusLine"].(map[string]any) + if !ok { + t.Fatalf("projection missing subagentStatusLine: %v", got) + } + if cmd, _ := sub["command"].(string); !strings.HasSuffix(cmd, " ctx-forward") { + t.Errorf("subagentStatusLine.command = %q, want ctx-forward suffix", cmd) + } +} + +// TestProjectionPolicyWinsOverFeed covers the merge contract: a policy +// that declares its own statusLine keeps it verbatim; the feed only adds +// keys the policy does not define. Falsification: if injection +// overwrites, the projected command is marvel's instead of the policy's. +func TestProjectionPolicyWinsOverFeed(t *testing.T) { + t.Parallel() + mgr, _ := projectionManager(t) + + m, err := api.ParseManifestBytes([]byte(feedWithPolicyManifest)) + if err != nil { + t.Fatalf("parse: %v", err) + } + if err := m.Apply(mgr.store); err != nil { + t.Fatalf("apply: %v", err) + } + key := seedFeedSession(t, mgr, "acme", "squad", "watcher", "squad-watcher-g1-0") + + if n := mgr.Reproject(); n != 1 { + t.Fatalf("Reproject changed = %d, want 1", n) + } + path := filepath.Join(mgr.ProjectionDir, strings.ReplaceAll(key, "/", "-")+".settings.json") + got := readProjection(t, path) + + sl := got["statusLine"].(map[string]any) + if cmd, _ := sl["command"].(string); cmd != "/usr/local/bin/my-statusline" { + t.Errorf("statusLine.command = %q, want the policy's own command (policy wins)", cmd) + } + if _, ok := got["subagentStatusLine"]; !ok { + t.Error("subagentStatusLine absent: feed should add keys the policy does not define") + } +}