diff --git a/.roborev.toml b/.roborev.toml index 7cfbdd69fe..101e3b9b32 100644 --- a/.roborev.toml +++ b/.roborev.toml @@ -1,10 +1,12 @@ review_guidelines = """ -agentsview is a single-user developer tool. Default mode binds to +agentsview defaults to a single-user developer tool. Local mode binds to 127.0.0.1. Optional managed Caddy proxy mode allows LAN access while keeping the backend on loopback. Optional remote access mode binds to 0.0.0.0 with bearer-token auth for use over secure tunnels (Tailscale, -SSH, reverse proxy with TLS). Not designed for multi-user or -internet-facing deployment. +SSH, reverse proxy with TLS). These legacy modes are not designed for +request-multiplexed multi-user deployment. Opt-in hosted raw processing has +a separate threat model: one configured tenant, schema, restricted role and +server instance, with authenticated uploads treated as untrusted input. Key assumptions reviewers MUST account for: @@ -15,7 +17,9 @@ Key assumptions reviewers MUST account for: bearer token is required for all API requests (including localhost, to prevent bypass via reverse proxy). An auth token is auto-generated at startup if missing. Do not flag missing auth on - local-only code paths. DO flag any path that lets the backend bind + local-only code paths. Hosted mode always requires authentication; + raw-sync routes use device credentials and scoped tokens separately + from the viewer bearer token. DO flag any path that lets the backend bind non-loopback in proxy mode, or missing subnet checks for non-loopback Caddy binds. @@ -37,8 +41,9 @@ Key assumptions reviewers MUST account for: 4. XSS: {@html renderMarkdown(...)} is safe — renderMarkdown() sanitizes via DOMPurify before returning HTML. -5. RATE LIMITING: Single-user tool. Do not flag missing rate limits - or concurrency caps. +5. RATE LIMITING: Do not flag missing rate limits or concurrency + caps in local/legacy single-user paths. Hosted raw ingestion and + derivation must bound body, materialization, output and worker resources. 6. CORS: corsMiddleware requires matching Origin for mutating requests. Allowed origins = loopback variants + public_url + @@ -46,25 +51,31 @@ Key assumptions reviewers MUST account for: token) allow the request origin. Do not flag as overly permissive unless origins outside the configured set are accepted. -7. INPUT VALIDATION: Body size limits not required — backend is - loopback-only; in proxy mode Caddy subnet filtering limits +7. INPUT VALIDATION: In local/legacy paths, body size limits are not + required — backend is loopback-only; in proxy mode Caddy subnet filtering limits clients to trusted hosts. In remote mode, bearer token gates - all API access. + all API access. Hosted raw authentication does not make source bytes + trusted; review bounded protocol and parser validation. 8. SESSION DATA: Displaying session contents (tool args, commands, paths) is the tool's purpose. The user owns these files. Do not - flag as sensitive data exposure. + flag as sensitive data exposure within that user's local/legacy access + boundary. Hosted reads must preserve tenant and public-identity isolation. -9. SUBPROCESS ENV: Agent CLI subprocesses intentionally inherit the - parent environment. Do not flag env var inheritance. +9. SUBPROCESS ENV: Local insight/capture agent CLIs intentionally inherit + the parent environment. Hosted raw parser children require an explicit + minimal environment and pre-runtime inherited-descriptor closure. -10. SESSION PARSING: Input files are from local agent CLIs, not - adversarial. Do not flag missing cycle detection, recursion - limits, or unreachable-node checks in DAG traversals. +10. SESSION PARSING: Local/legacy input files are user-owned. Do not flag + missing cycle detection, recursion limits or unreachable-node checks + there. Hosted raw input is untrusted; review fail-closed filesystem, + network, process and resource isolation before parsing. -11. WRITE ATOMICITY: Per-session transactions only. Full resync - recovers partial state. Do not flag non-atomic multi-session - writes. +11. WRITE ATOMICITY: Local/legacy writes use per-session transactions; + full resync recovers partial state. Hosted publication must atomically + fence selection and lease generations, normalized rows, provenance, + public identity, curation, outbox and job outcome. Partial sources must + retain unresolved proof and retry finitely. 12. TOCTOU ON LOCAL FILES: Files in ~/.agentsview/ are user-owned. An attacker with home-directory access already has equivalent @@ -132,7 +143,11 @@ Key assumptions reviewers MUST account for: go.mod toolchain version before flagging unfamiliar language features. -Do NOT flag issues that only apply to public-facing, multi-tenant, -or internet-exposed services. Focus on bugs, logic errors, data -corruption risks, and code quality issues. +For local/legacy modes, do not impose requirements that apply only to +public-facing multi-tenant services. For opt-in hosted raw processing, +review its actual tenant/role/catalog boundary, immutable source provenance, +source-selection and lease fences, bounded isolated parsing, public-ID +ambiguity, curation survival and partial-source retry semantics. Do not +assume request-multiplexed tenancy. Focus on bugs, data corruption, boundary +violations and maintainability within each mode's supported contract. """ diff --git a/README.md b/README.md index b0e1bfeab0..df35d71d3b 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,7 @@ +--- +last_edited: 2026-09-11 +--- + # agentsview Browse, search, and track costs across all your AI coding agents. One binary, no @@ -557,6 +561,11 @@ agentsview pg status --all # show status for every configured PG target agentsview pg serve # serve web UI from the default PG target (read-only) ``` +Opt-in [hosted raw processing](docs/hosted-raw-sync.md) lets `pg serve` parse +uploaded sources directly into PostgreSQL. It requires explicit tenant/schema +provisioning, a restricted runtime role, authentication and Linux isolation. +Ordinary PG serving remains read-only; `pg push` refuses hosted-owned schemas. + Single-target configs still use the legacy `[pg]` block. To manage more than one PostgreSQL destination, define named `[pg.NAME]` blocks and set `default_pg` when more than one target exists: @@ -576,7 +585,9 @@ exclude_projects = ["scratch"] Named target names are normalized case-insensitively. `all`, `local`, and the legacy `[pg]` field names `url`, `schema`, `machine_name`, `allow_insecure`, -`projects`, and `exclude_projects` cannot be used for `[pg.NAME]`. +`projects`, `exclude_projects`, `raw_tenant`, `raw_derivation`, +`raw_poll_seconds`, `raw_attempt_seconds`, and `raw_max_attempts` cannot be used +for `[pg.NAME]`. `AGENTSVIEW_PG_URL`, `AGENTSVIEW_PG_SCHEMA`, and `AGENTSVIEW_PG_MACHINE` still work, but in named-target mode they apply only to the effective default target. diff --git a/cmd/agentsview/cli.go b/cmd/agentsview/cli.go index 51bf5055f5..1a39653052 100644 --- a/cmd/agentsview/cli.go +++ b/cmd/agentsview/cli.go @@ -651,6 +651,8 @@ func newPGCommand() *cobra.Command { cmd.AddCommand(newPGServeCommand()) cmd.AddCommand(newPGVectorsCommand()) cmd.AddCommand(newPGServiceCommand()) + cmd.AddCommand(newPGHostedProvisionCommand()) + cmd.AddCommand(newPGRawReparseCommand()) return cmd } diff --git a/cmd/agentsview/main.go b/cmd/agentsview/main.go index 9eb9fcd1c3..0329d35308 100644 --- a/cmd/agentsview/main.go +++ b/cmd/agentsview/main.go @@ -24,6 +24,7 @@ import ( "go.kenn.io/agentsview/internal/config" "go.kenn.io/agentsview/internal/db" "go.kenn.io/agentsview/internal/parser" + "go.kenn.io/agentsview/internal/rawderive" "go.kenn.io/agentsview/internal/recall/extract" "go.kenn.io/agentsview/internal/remotesync" "go.kenn.io/agentsview/internal/secrets" @@ -60,6 +61,9 @@ const ( ) func main() { + if handled, code := rawderive.RunParserChild(os.Args[1:]); handled { + os.Exit(code) + } // Turn on the agentsview-test-fixture deny-list before any scan // runs. The secrets package keeps the filter off by default so unit // tests in this repo (which use the same random-looking fixtures diff --git a/cmd/agentsview/pg.go b/cmd/agentsview/pg.go index 02ce875a42..6639519811 100644 --- a/cmd/agentsview/pg.go +++ b/cmd/agentsview/pg.go @@ -114,6 +114,11 @@ func runPGPush( return err } + for _, target := range targets { + if target.PG.RawTenant != "" || target.PG.RawDerivation { + return errors.New("pg push cannot mutate a hosted-owned projection") + } + } applyClassifierConfig(appCfg) ctx, stop := signal.NotifyContext( context.Background(), os.Interrupt, @@ -172,6 +177,9 @@ func runPGPushTarget( if err != nil { return err } + if target.PG.RawTenant != "" || target.PG.RawDerivation { + return errors.New("pg push cannot mutate a hosted-owned projection") + } if target.PG.URL == "" { return fmt.Errorf("url not configured") } @@ -479,11 +487,12 @@ func loadPGServeConfig(cmd *cobra.Command) (config.Config, string, error) { } type pgServeStartup struct { - cfg config.Config - ctx context.Context - rtOpts serveRuntimeOptions - srv *server.Server - cleanup func() + cfg config.Config + ctx context.Context + rtOpts serveRuntimeOptions + srv *server.Server + cleanup func() + startWorker func() } var preparePGServe = preparePGServeImpl @@ -501,6 +510,12 @@ func preparePGServeImpl(appCfg config.Config, basePath string) (pgServeStartup, return pgServeStartup{}, errors.New("pg serve: url not configured") } + if err := pgCfg.ValidateRawDerivation(appCfg.RequireAuth); err != nil { + return pgServeStartup{}, err + } + if pgCfg.RawTenant != "" { + return prepareHostedPGServe(appCfg, pgCfg, basePath) + } applyClassifierConfig(appCfg) store, err := postgres.NewStore( pgCfg.URL, pgCfg.Schema, pgCfg.AllowInsecure, @@ -632,8 +647,16 @@ func runPGServe(appCfg config.Config, basePath string) { if err != nil { fatal("%v", err) } + if err = runPreparedPGServe(startup); err != nil { + fatal("%v", err) + } +} + +// Return through cleanup before the outer CLI may call os.Exit, including +// readiness failures and unexpected server/proxy exits. +func runPreparedPGServe(startup pgServeStartup) error { defer startup.cleanup() - appCfg = startup.cfg + appCfg := startup.cfg ctx := startup.ctx rtOpts := startup.rtOpts srv := startup.srv @@ -646,11 +669,14 @@ func runPGServe(appCfg config.Config, basePath string) { ) if err != nil { if errors.Is(err, context.Canceled) { - return + return nil } - fatal("pg serve: %v", err) + return fmt.Errorf("pg serve: %w", err) } + if startup.startWorker != nil { + startup.startWorker() + } // Write the kit runtime record so CLI commands can discover this // daemon. ReadOnly=true marks it as pg serve (read-only) // so clients can select an appropriate transport. @@ -677,8 +703,9 @@ func runPGServe(appCfg config.Config, basePath string) { } if err := waitForServerRuntime(ctx, srv, rt); err != nil { - fatal("pg serve: %v", err) + return fmt.Errorf("pg serve: %w", err) } + return nil } func writePGServeRuntimeRecord(rt *serveRuntime) bool { diff --git a/cmd/agentsview/pg_raw_derive.go b/cmd/agentsview/pg_raw_derive.go new file mode 100644 index 0000000000..e2a239b7c6 --- /dev/null +++ b/cmd/agentsview/pg_raw_derive.go @@ -0,0 +1,276 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "os" + "os/signal" + "strings" + "sync" + "syscall" + "time" + + "github.com/spf13/cobra" + "go.kenn.io/agentsview/internal/config" + "go.kenn.io/agentsview/internal/db" + "go.kenn.io/agentsview/internal/ingest" + "go.kenn.io/agentsview/internal/postgres" + "go.kenn.io/agentsview/internal/rawderive" + "go.kenn.io/agentsview/internal/rawsync" + "go.kenn.io/agentsview/internal/server" +) + +func rawProcessingVersion() string { return fmt.Sprintf("parser-data-%d", db.CurrentDataVersion()) } + +// pgRawRuntime owns one bounded sequential worker/maintenance loop. Stop joins +// all materialization/parser work before custody and database owners may close. +type pgRawRuntime struct { + mu sync.Mutex + stopped, started bool + cancel context.CancelFunc + done chan struct{} + ctx context.Context + batch func(context.Context) + interval time.Duration +} + +func newPGRawRuntime(parent context.Context, interval time.Duration, batch func(context.Context)) *pgRawRuntime { + ctx, cancel := context.WithCancel(parent) + return &pgRawRuntime{ctx: ctx, cancel: cancel, done: make(chan struct{}), batch: batch, interval: interval} +} +func (r *pgRawRuntime) Start() { + r.mu.Lock() + defer r.mu.Unlock() + if r.started || r.stopped { + return + } + r.started = true + go func() { + defer close(r.done) + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + for { + if r.ctx.Err() != nil { + return + } + r.batch(r.ctx) + select { + case <-r.ctx.Done(): + return + case <-ticker.C: + } + } + }() +} +func (r *pgRawRuntime) Stop() { + r.mu.Lock() + r.stopped = true + r.cancel() + started := r.started + r.mu.Unlock() + if started { + <-r.done + } +} + +func prepareHostedPGServe(app config.Config, pg config.PGConfig, basePath string) (pgServeStartup, error) { + if err := pg.ValidateRawDerivation(app.RequireAuth); err != nil { + return pgServeStartup{}, err + } + applyClassifierConfig(app) + store, err := postgres.NewHostedStore(pg.URL, pg.Schema, pg.RawTenant, pg.AllowInsecure) + if err != nil { + return pgServeStartup{}, err + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + var runtime *pgRawRuntime + var closeUploads func() error + custody := &pgRawSyncCustody{dataDir: app.DataDir, tenant: pg.RawTenant, limits: rawsync.DefaultManifestLimits(), version: rawProcessingVersion()} + var once sync.Once + cleanup := func() { + once.Do(func() { + stop() + if runtime != nil { + runtime.Stop() + } + if closeUploads != nil { + if err := closeUploads(); err != nil { + log.Print("hosted raw upload cleanup failed") + } + } + if err := custody.Close(); err != nil { + log.Print("hosted raw custody cleanup failed") + } + if err := store.Close(); err != nil { + log.Print("hosted raw database cleanup failed") + } + }) + } + fail := func(err error) (pgServeStartup, error) { cleanup(); return pgServeStartup{}, err } + if err = applyRequiredCursorSecret(store, app); err != nil { + return fail(err) + } + store.SetCustomPricing(app.CustomModelPricing) + if err = postgres.CheckHostedRuntimeWritable(ctx, store.DB(), pg.Schema, app.ArchiveContent); err != nil { + return fail(err) + } + metadata, err := postgres.NewHostedRawIngestStore(store.DB(), pg.RawTenant, custody.version) + if err != nil { + return fail(err) + } + custody.metadata = metadata + authStore, err := postgres.NewTenantRawDeviceAuthStore(store.DB(), pg.RawTenant) + if err != nil { + return fail(err) + } + auth, err := rawsync.NewDeviceAuthService(authStore, pgRawSyncTokenTTL) + if err != nil { + return fail(err) + } + if pg.RawDerivation { + poll, attempt, maxAttempts := pg.RawWorkerBounds() + isolated, err := rawderive.NewSubprocessParser(time.Duration(attempt) * time.Second) + if err != nil { + return fail(err) + } + if err = isolated.Preflight(ctx); err != nil { + return fail(err) + } + retry := rawderive.RetryPolicy{Base: time.Second, Maximum: time.Minute, MaxAttempts: maxAttempts} + sink, err := postgres.NewRawProjectionStore(store.DB(), hostedRawProjectionOptions(app, pg.RawTenant, retry)) + if err != nil { + return fail(err) + } + worker, err := rawderive.NewWorker(rawderive.WorkerConfig{ + Queue: metadata, Manifests: rawderive.ManifestLoader{Store: custody, Limits: custody.limits}, + Materializer: rawderive.Materializer{Store: custody, BaseDir: os.TempDir(), MaxTotalBytes: 512 << 20}, Parser: isolated, Projection: sink, + Owner: fmt.Sprintf("hosted-%d", os.Getpid()), BatchSize: 1, LeaseDuration: time.Minute, HeartbeatInterval: 10 * time.Second, AttemptTimeout: time.Duration(attempt) * time.Second, + RetryBase: retry.Base, RetryMax: retry.Maximum, MaxAttempts: retry.MaxAttempts, + }) + if err != nil { + return fail(err) + } + runtime = newPGRawRuntime(ctx, time.Duration(poll)*time.Second, func(ctx context.Context) { + result, err := worker.RunBatch(ctx) + if err != nil && ctx.Err() == nil { + log.Printf("hosted raw worker: claimed=%d succeeded=%d retried=%d failed=%d; batch failed", result.Claimed, result.Succeeded, result.Retried, result.Failed) + } + if ctx.Err() != nil { + return + } + maintenance, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if _, err = sink.SettlePendingSignals(maintenance, 64); err != nil && ctx.Err() == nil { + log.Print("hosted raw pending-signal maintenance failed") + } + }) + } + rtOpts := serveRuntimeOptions{Mode: "pg-serve", RequestedPort: app.Port} + app, err = prepareServeRuntimeConfig(app, rtOpts) + if err != nil { + return fail(err) + } + uploadOption, closeUploadStore, err := preparePGRawSyncUploads(app.DataDir, store.DB(), custody) + if err != nil { + return fail(err) + } + closeUploads = closeUploadStore + opts := []server.Option{server.WithVersion(server.VersionInfo{Version: version, Commit: commit, BuildDate: buildDate, ReadOnly: true}), server.WithDataDir(app.DataDir), server.WithBaseContext(ctx), server.WithRawSyncServices(auth, custody), server.WithRawSyncTenant(pg.RawTenant), uploadOption} + if basePath != "" { + opts = append(opts, server.WithBasePath(basePath)) + } + startup := pgServeStartup{cfg: app, ctx: ctx, rtOpts: rtOpts, srv: server.New(app, store, nil, opts...), cleanup: cleanup} + if runtime != nil { + startup.startWorker = runtime.Start + } + return startup, nil +} + +func newPGHostedProvisionCommand() *cobra.Command { + return &cobra.Command{Use: "hosted-provision [target]", Short: "Explicitly provision one hosted tenant schema using the configured owner connection", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { + cfg, err := config.LoadMinimal() + if err != nil { + return err + } + target := "" + if len(args) > 0 { + target = args[0] + } + pg, err := cfg.ResolvePGTarget(target) + if err != nil { + return err + } + if pg.RawTenant == "" { + return errors.New("hosted-provision requires raw_tenant in the selected PG target") + } + applyClassifierConfig(cfg) + database, err := postgres.Open(pg.URL, pg.Schema, pg.AllowInsecure) + if err != nil { + return errors.New("opening hosted owner connection failed") + } + defer database.Close() + if err = postgres.EnsureHostedTenant(cmd.Context(), database, pg.Schema, pg.RawTenant); err != nil { + return fmt.Errorf("hosted provisioning failed: %w", err) + } + fmt.Fprintln(cmd.OutOrStdout(), "Hosted tenant schema provisioned. Use a restricted tenant runtime connection to serve.") + return nil + }} +} +func newPGRawReparseCommand() *cobra.Command { + var runID string + var batch int + cmd := &cobra.Command{Use: "raw-reparse [target]", Short: "Schedule one bounded, resumable batch of current raw heads for this parser version", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { + if runID == "" { + return errors.New("raw-reparse requires --run-id for its durable checkpoint") + } + cfg, err := config.LoadMinimal() + if err != nil { + return err + } + target := "" + if len(args) > 0 { + target = args[0] + } + pg, err := cfg.ResolvePGTarget(target) + if err != nil { + return err + } + if err = pg.ValidateRawDerivation(cfg.RequireAuth); err != nil { + return err + } + if pg.RawTenant == "" { + return errors.New("raw-reparse requires raw_tenant") + } + store, err := postgres.NewHostedStore(pg.URL, pg.Schema, pg.RawTenant, pg.AllowInsecure) + if err != nil { + return err + } + defer store.Close() + sink, err := postgres.NewRawProjectionStore(store.DB(), postgres.RawProjectionOptions{Tenant: pg.RawTenant}) + if err != nil { + return err + } + result, err := sink.ScheduleCurrentHeads(cmd.Context(), runID, rawProcessingVersion(), batch) + if err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), "Selected %d current heads; complete=%t\n", result.Selected, result.Done) + return nil + }} + cmd.Flags().StringVar(&runID, "run-id", "", "Durable rollout checkpoint ID; reuse to resume") + cmd.Flags().IntVar(&batch, "batch-size", 64, "Maximum current heads in this invocation (1-256)") + return cmd +} + +func hostedRawProjectionOptions(app config.Config, tenant string, retry rawderive.RetryPolicy) postgres.RawProjectionOptions { + blocked := make(map[string]bool, len(app.ResultContentBlockedCategories)) + for _, category := range app.ResultContentBlockedCategories { + category = strings.TrimSpace(category) + if category != "" { + blocked[strings.ToUpper(category[:1])+strings.ToLower(category[1:])] = true + } + } + return postgres.RawProjectionOptions{Tenant: tenant, RetryPolicy: retry, Content: ingest.ContentOptions{ArchiveContent: app.ArchiveContent, ToolResultImages: app.ToolResultImages, BlockedResultCategories: blocked}} +} diff --git a/cmd/agentsview/pg_raw_derive_pgtest_test.go b/cmd/agentsview/pg_raw_derive_pgtest_test.go new file mode 100644 index 0000000000..093facabe0 --- /dev/null +++ b/cmd/agentsview/pg_raw_derive_pgtest_test.go @@ -0,0 +1,226 @@ +//go:build pgtest + +package main + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/config" + "go.kenn.io/agentsview/internal/parser" + "go.kenn.io/agentsview/internal/postgres" + "go.kenn.io/agentsview/internal/rawderive" + "go.kenn.io/agentsview/internal/rawsync" +) + +func hostedRuntimeConfig(t *testing.T) (config.Config, *sql.DB) { + t.Helper() + dsn := os.Getenv("TEST_PG_URL") + if dsn == "" { + t.Skip("TEST_PG_URL not configured") + } + nonce := make([]byte, 16) + _, err := rand.Read(nonce) + require.NoError(t, err) + suffix := hex.EncodeToString(nonce) + schema := "runtime_test_" + suffix + role := schema + "_role" + admin, err := postgres.Open(dsn, schema, false) + require.NoError(t, err) + t.Cleanup(func() { + _, err := admin.ExecContext(context.Background(), `DROP SCHEMA "`+schema+`" CASCADE;DROP ROLE "`+role+`"`) + assert.NoError(t, err) + assert.NoError(t, admin.Close()) + }) + require.NoError(t, postgres.EnsureHostedTenant(t.Context(), admin, schema, "tenant-runtime")) + password := make([]byte, 32) + _, err = rand.Read(password) + require.NoError(t, err) + _, err = admin.Exec(`CREATE ROLE "` + role + `" LOGIN NOSUPERUSER NOBYPASSRLS NOCREATEDB NOCREATEROLE NOINHERIT PASSWORD '` + hex.EncodeToString(password) + `';GRANT USAGE ON SCHEMA "` + schema + `" TO "` + role + `";GRANT SELECT,INSERT,UPDATE,DELETE ON ALL TABLES IN SCHEMA "` + schema + `" TO "` + role + `";GRANT USAGE ON ALL SEQUENCES IN SCHEMA "` + schema + `" TO "` + role + `"`) + require.NoError(t, err) + u, err := url.Parse(dsn) + require.NoError(t, err) + u.User = url.UserPassword(role, hex.EncodeToString(password)) + cfg := config.Config{DataDir: t.TempDir(), Host: "127.0.0.1", Port: 0, RequireAuth: true, AuthToken: "synthetic-test-token", CursorSecret: "c3ludGhldGljLWN1cnNvci1zZWNyZXQ=", NoBrowser: true, PG: config.PGConfig{URL: u.String(), Schema: schema, RawTenant: "tenant-runtime"}} + return cfg, admin +} + +func TestHostedRuntimePreparationOffRetainsPublicReads(t *testing.T) { + cfg, _ := hostedRuntimeConfig(t) + // An occupied custody path makes any eager repository open fail decisively. + require.NoError(t, os.WriteFile(filepath.Join(cfg.DataDir, pgRawSyncDataDirectory), []byte("occupied"), 0600)) + startup, err := preparePGServeImpl(cfg, "") + require.NoError(t, err) + require.Nil(t, startup.startWorker) + startup.cleanup() + startup.cleanup() + store, closeStore, err := openPGReadStore(cfg, cfg.PG) + require.NoError(t, err) + defer closeStore() + _, ok := store.(*postgres.HostedStore) + require.True(t, ok) + cfg.PG.RawTenant = "" + _, err = preparePGServeImpl(cfg, "") + require.ErrorContains(t, err, "raw_tenant") +} +func TestHostedRuntimePreparationSandboxAndIdle(t *testing.T) { + cfg, _ := hostedRuntimeConfig(t) + cfg.PG.RawDerivation = true + p, err := rawderive.NewSubprocessParser(5 * time.Second) + require.NoError(t, err) + supported := p.Preflight(t.Context()) == nil + if !supported && os.Getenv("RAW_SANDBOX_REQUIRED") == "1" { + t.Fatal("mandatory sandbox unavailable") + } + require.NoError(t, os.WriteFile(filepath.Join(cfg.DataDir, pgRawSyncDataDirectory), []byte("occupied"), 0600)) + startup, err := preparePGServeImpl(cfg, "") + if !supported { + require.ErrorIs(t, err, rawderive.ErrSandboxUnavailable) + return + } + require.NoError(t, err) + defer startup.cleanup() + require.NotNil(t, startup.startWorker) + startup.startWorker() + // This cancellation joins the real empty queue/maintenance pass. Occupied + // custody proves startup and empty work have no repository dependency. + startup.cleanup() +} + +// This gate executes the actual hosted startup worker and parser child. It is +// mandatory on the Linux isolation runner; the local restricted host skips it. +func TestHostedRuntimeDerivesAcceptedSource(t *testing.T) { + p, err := rawderive.NewSubprocessParser(5 * time.Second) + require.NoError(t, err) + if err = p.Preflight(t.Context()); err != nil { + if os.Getenv("RAW_SANDBOX_REQUIRED") == "1" { + t.Fatal(err) + } + t.Skip("kernel isolation unavailable") + } + cfg, _ := hostedRuntimeConfig(t) + cfg.PG.RawDerivation = true + cfg.PG.RawPollSeconds = 1 + startup, err := preparePGServeImpl(cfg, "") + require.NoError(t, err) + defer startup.cleanup() + database, err := postgres.OpenHosted(cfg.PG.URL, cfg.PG.Schema, cfg.PG.RawTenant, false) + require.NoError(t, err) + defer database.Close() + authStore, err := postgres.NewTenantRawDeviceAuthStore(database, cfg.PG.RawTenant) + require.NoError(t, err) + auth, err := rawsync.NewDeviceAuthService(authStore, time.Minute) + require.NoError(t, err) + enrolled, err := auth.EnrollDevice(t.Context(), cfg.PG.RawTenant, "synthetic device") + require.NoError(t, err) + request := func(method, path, token, kind string, body []byte) *httptest.ResponseRecorder { + req := httptest.NewRequest(method, "http://127.0.0.1"+path, bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", kind) + req.Header.Set("X-AgentsView-Device-ID", enrolled.Identity.DeviceID) + req.Header.Set("Upload-Offset", "0") + rec := httptest.NewRecorder() + startup.srv.Handler().ServeHTTP(rec, req) + return rec + } + tokenResponse := request("POST", "/api/v1/raw-sync/tokens", enrolled.Credential, "application/json", []byte(`{"scopes":["upload","commit"]}`)) + require.Equal(t, 200, tokenResponse.Code) + var token struct{ Token string } + require.NoError(t, json.Unmarshal(tokenResponse.Body.Bytes(), &token)) + body := hostedRuntimeClaudeFixture() + digest := sha256.Sum256(body) + ref := rawsync.ObjectRef{SHA256: hex.EncodeToString(digest[:]), Length: int64(len(body))} + encoded, err := json.Marshal(map[string]any{"provider": "claude", "object": ref}) + require.NoError(t, err) + uploadResponse := request("POST", "/api/v1/raw-sync/uploads", token.Token, "application/json", encoded) + require.Contains(t, []int{200, 201}, uploadResponse.Code) + var upload struct { + UploadID string `json:"upload_id"` + } + require.NoError(t, json.Unmarshal(uploadResponse.Body.Bytes(), &upload)) + require.NotEmpty(t, upload.UploadID) + appended := request("PATCH", "/api/v1/raw-sync/uploads/"+upload.UploadID, token.Token, "application/octet-stream", body) + require.Equal(t, 200, appended.Code) + manifest := rawsync.Manifest{SchemaVersion: rawsync.ManifestSchemaVersion, Provider: parser.AgentClaude, ConfiguredRootID: "root", SourceKey: "/canonical/project/runtime-session.jsonl", CaptureID: "runtime-capture", CapturedAt: time.Now().UTC(), Kind: rawsync.ManifestSnapshot, Entries: []rawsync.Entry{{Path: "project/runtime-session.jsonl", Type: "file", Length: ref.Length, Objects: []rawsync.ObjectRef{ref}}}} + encoded, err = json.Marshal(manifest) + require.NoError(t, err) + accepted := request("POST", "/api/v1/raw-sync/manifests", token.Token, "application/json", encoded) + require.Equal(t, 200, accepted.Code, accepted.Body.String()) + var attempts, generation int + require.NoError(t, database.QueryRow(`SELECT attempt_count,projection_generation FROM raw_ingest_jobs`).Scan(&attempts, &generation)) + assert.Zero(t, attempts) + assert.Equal(t, 1, generation) + ctx, cancel := context.WithCancel(startup.ctx) + startup.ctx = ctx + defer cancel() + finished := make(chan error, 1) + go func() { finished <- runPreparedPGServe(startup) }() + defer func() { + cancel() + select { + case err := <-finished: + require.NoError(t, err) + case <-time.After(10 * time.Second): + t.Error("hosted runtime did not join") + } + }() + require.Eventually(t, func() bool { + var state string + err := database.QueryRow(`SELECT state FROM raw_ingest_jobs`).Scan(&state) + return err == nil && state == "complete" + }, 20*time.Second, 50*time.Millisecond) + store, err := postgres.NewHostedStore(cfg.PG.URL, cfg.PG.Schema, cfg.PG.RawTenant, false) + require.NoError(t, err) + defer store.Close() + messages, err := store.GetMessages(t.Context(), "runtime-session", 0, 10, true) + require.NoError(t, err) + require.Len(t, messages, 2) + assert.Equal(t, "hello runtime", messages[0].Content) + assert.Equal(t, "hello viewer", messages[1].Content) +} + +func TestHostedRuntimeRejectsMissingPublicationPrivilegesBeforeReadiness(t *testing.T) { + cfg, admin := hostedRuntimeConfig(t) + endpoint, err := url.Parse(cfg.PG.URL) + require.NoError(t, err) + role := endpoint.User.Username() + control, err := preparePGServeImpl(cfg, "") + require.NoError(t, err) + control.cleanup() + cfg.PG.RawDerivation = true + for _, tc := range []struct{ object, privilege, kind string }{ + {"session_sources", "DELETE", "TABLE"}, {"raw_session_links", "DELETE", "TABLE"}, {"tool_result_events", "INSERT", "TABLE"}, {"usage_events", "INSERT", "TABLE"}, {"secret_findings", "INSERT", "TABLE"}, {"starred_sessions", "INSERT", "TABLE"}, {"pinned_messages", "UPDATE", "TABLE"}, + {"tool_calls_id_seq", "USAGE", "SEQUENCE"}, {"tool_result_events_id_seq", "USAGE", "SEQUENCE"}, {"usage_events_id_seq", "USAGE", "SEQUENCE"}, {"pinned_messages_id_seq", "USAGE", "SEQUENCE"}, + } { + t.Run(tc.object+"_"+tc.privilege, func(t *testing.T) { + _, err := admin.Exec(`REVOKE ` + tc.privilege + ` ON ` + tc.kind + ` "` + tc.object + `" FROM "` + role + `"`) + require.NoError(t, err) + defer func() { + _, err := admin.Exec(`GRANT ` + tc.privilege + ` ON ` + tc.kind + ` "` + tc.object + `" TO "` + role + `"`) + require.NoError(t, err) + }() + startup, err := preparePGServeImpl(cfg, "") + if err == nil { + startup.cleanup() + } + require.ErrorContains(t, err, "privileges") + require.Nil(t, startup.startWorker) + var attempts int + require.NoError(t, admin.QueryRow(`SELECT COALESCE(sum(attempt_count),0) FROM raw_ingest_jobs`).Scan(&attempts)) + assert.Zero(t, attempts) + }) + } +} diff --git a/cmd/agentsview/pg_raw_derive_test.go b/cmd/agentsview/pg_raw_derive_test.go new file mode 100644 index 0000000000..edfae4c851 --- /dev/null +++ b/cmd/agentsview/pg_raw_derive_test.go @@ -0,0 +1,166 @@ +package main + +import ( + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/config" + "go.kenn.io/agentsview/internal/ingest" + "go.kenn.io/agentsview/internal/parser" + "go.kenn.io/agentsview/internal/rawderive" + "go.kenn.io/agentsview/internal/server" +) + +func TestPGRawRuntimeChild(t *testing.T) { + if os.Getenv("RAW_RUNTIME_CHILD") != "1" { + return + } + for { + time.Sleep(time.Hour) + } +} + +func TestPGRawRuntimeReadinessCancellationJoinsChild(t *testing.T) { + var calls atomic.Int32 + entered := make(chan *exec.Cmd, 1) + joined := make(chan struct{}) + r := newPGRawRuntime(t.Context(), time.Hour, func(ctx context.Context) { + calls.Add(1) + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestPGRawRuntimeChild$") + cmd.Env = []string{"RAW_RUNTIME_CHILD=1"} + if err := cmd.Start(); err != nil { + entered <- nil + return + } + entered <- cmd + _ = cmd.Wait() + close(joined) + }) + assert.Zero(t, calls.Load(), "preparation must not start claims") + r.Start() + r.Start() + var child *exec.Cmd + select { + case child = <-entered: + case <-time.After(3 * time.Second): + t.Fatal("worker did not start") + } + require.NotNil(t, child) + r.Stop() + select { + case <-joined: + default: + t.Fatal("Stop returned before child Wait") + } + require.NotNil(t, child.ProcessState) + r.Start() + r.Stop() + assert.EqualValues(t, 1, calls.Load()) +} +func TestPGRawRuntimeClosedBeforeReadinessNeverStarts(t *testing.T) { + var calls atomic.Int32 + r := newPGRawRuntime(t.Context(), time.Second, func(context.Context) { calls.Add(1) }) + r.Stop() + r.Start() + r.Stop() + assert.Zero(t, calls.Load()) +} + +func TestMain(m *testing.M) { + if handled, code := rawderive.RunParserChild(os.Args[1:]); handled { + os.Exit(code) + } + os.Exit(m.Run()) +} + +func TestPGRawRuntimeReadinessFailureCleansBeforeReturn(t *testing.T) { + var cleaned, started atomic.Bool + cfg := config.Config{DataDir: t.TempDir(), Host: "invalid host", Port: 12345, NoBrowser: true} + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + startup := pgServeStartup{cfg: cfg, ctx: ctx, srv: server.New(cfg, nil, nil), rtOpts: serveRuntimeOptions{Mode: "pg-serve", RequestedPort: cfg.Port}, cleanup: func() { cleaned.Store(true) }, startWorker: func() { started.Store(true) }} + require.Error(t, runPreparedPGServe(startup)) + assert.True(t, cleaned.Load()) + assert.False(t, started.Load()) +} + +func TestHostedProjectionConfiguredContentPolicies(t *testing.T) { + parsed := parser.ParseResult{Session: parser.ParsedSession{ID: "policy-session", Agent: parser.AgentClaude}, Messages: []parser.ParsedMessage{ + {Ordinal: 0, Role: parser.RoleAssistant, Content: "checking", ToolCalls: []parser.ParsedToolCall{{ToolUseID: "read", ToolName: "Read", Category: "Read", InputJSON: `{"path":"file"}`}, {ToolUseID: "bash", ToolName: "Bash", Category: "Bash", InputJSON: `{"command":"true"}`}}}, + {Ordinal: 1, Role: parser.RoleUser, ToolResults: []parser.ParsedToolResult{{ToolUseID: "read", ContentRaw: `"private result"`, ContentLength: 14}, {ToolUseID: "bash", ContentRaw: `[{"type":"text","text":"before"},{"type":"input_image","image_url":"data:image/png;base64,AAEC"}]`}}}, + }} + imageContent, err := json.Marshal(`[{"type":"text","text":"before"},{"type":"input_image","image_url":"data:image/png;base64,AAEC"}]`) + require.NoError(t, err) + parsed.Messages[1].ToolResults[1].ContentRaw = string(imageContent) + for _, mode := range []string{"full", "blocked", "images", "transcripts"} { + t.Run(mode, func(t *testing.T) { + app := config.Config{} + switch mode { + case "blocked": + app.ResultContentBlockedCategories = []string{" rEaD ", " "} + case "images": + app.ToolResultImages = config.ToolResultImagesDrop + case "transcripts": + app.ArchiveContent = config.ArchiveContentTranscripts + } + options := hostedRawProjectionOptions(app, "tenant", rawderive.RetryPolicy{}).Content + candidate, err := ingest.PrepareCandidate(t.Context(), parsed, options) + require.NoError(t, err) + prepared, err := ingest.Finalize(t.Context(), candidate, options) + require.NoError(t, err) + require.Len(t, prepared.Messages, 1) + require.Len(t, prepared.Messages[0].ToolCalls, 2) + read, bash := prepared.Messages[0].ToolCalls[0], prepared.Messages[0].ToolCalls[1] + if mode == "blocked" || mode == "transcripts" { + assert.Empty(t, read.ResultContent) + } else { + assert.Equal(t, "private result", read.ResultContent) + } + if mode == "images" || mode == "transcripts" { + assert.NotContains(t, bash.ResultContent, "base64,AAEC") + } else { + assert.Contains(t, bash.ResultContent, "base64,AAEC") + } + if mode == "transcripts" { + assert.Empty(t, bash.InputJSON) + } else { + assert.Equal(t, `{"command":"true"}`, bash.InputJSON) + } + }) + } +} + +func hostedRuntimeClaudeFixture() []byte { + return []byte(`{"type":"user","timestamp":"2026-08-13T12:00:00Z","uuid":"u1","sessionId":"runtime-session","message":{"content":"hello runtime"},"cwd":"/work/project"}` + "\n" + `{"type":"assistant","timestamp":"2026-08-13T12:00:01Z","uuid":"a1","parentUuid":"u1","sessionId":"runtime-session","message":{"content":"hello viewer"}}` + "\n") +} + +func TestHostedRuntimeClaudeFixtureHasTwoPreparedMessages(t *testing.T) { + root := t.TempDir() + path := filepath.Join(root, "project", "runtime-session.jsonl") + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0700)) + require.NoError(t, os.WriteFile(path, hostedRuntimeClaudeFixture(), 0400)) + factory, ok := parser.ProviderFactoryByType(parser.AgentClaude) + require.True(t, ok) + provider := factory.NewProvider(parser.ProviderConfig{Roots: []string{root}, Machine: "hosted"}) + outcome, err := provider.Parse(parser.WithoutFilesystemProjectDiscovery(t.Context()), parser.ParseRequest{Source: parser.SourceRef{Provider: parser.AgentClaude, DisplayPath: path, Key: path, ProjectHint: "project"}}) + require.NoError(t, err) + require.Len(t, outcome.Results, 1) + candidate, err := ingest.PrepareCandidate(t.Context(), outcome.Results[0].Result, ingest.ContentOptions{}) + require.NoError(t, err) + prepared, err := ingest.Finalize(t.Context(), candidate, ingest.ContentOptions{}) + require.NoError(t, err) + require.Len(t, prepared.Messages, 2) + assert.Equal(t, "runtime-session", prepared.Session.ID) + assert.Equal(t, "user", prepared.Messages[0].Role) + assert.Equal(t, "hello runtime", prepared.Messages[0].Content) + assert.Equal(t, "assistant", prepared.Messages[1].Role) + assert.Equal(t, "hello viewer", prepared.Messages[1].Content) +} diff --git a/cmd/agentsview/pg_raw_parity_pgtest_test.go b/cmd/agentsview/pg_raw_parity_pgtest_test.go new file mode 100644 index 0000000000..c06dbf23e4 --- /dev/null +++ b/cmd/agentsview/pg_raw_parity_pgtest_test.go @@ -0,0 +1,436 @@ +//go:build pgtest + +package main + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "net/http/httptest" + "os" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/config" + "go.kenn.io/agentsview/internal/db" + "go.kenn.io/agentsview/internal/parser" + "go.kenn.io/agentsview/internal/postgres" + "go.kenn.io/agentsview/internal/rawcapture" + "go.kenn.io/agentsview/internal/rawcheckpoint" + "go.kenn.io/agentsview/internal/rawderive" + "go.kenn.io/agentsview/internal/rawsync" + "go.kenn.io/agentsview/internal/rawtest" +) + +type hostedCaptureClient struct { + checkpoint *rawcheckpoint.Store + provider parser.Provider + device, token string + startup *pgServeStartup + last rawsync.Manifest +} + +func requireHostedSandbox(t *testing.T) { + t.Helper() + p, err := rawderive.NewSubprocessParser(5 * time.Second) + require.NoError(t, err) + if err = p.Preflight(t.Context()); err != nil { + if os.Getenv("RAW_SANDBOX_REQUIRED") == "1" { + t.Fatal(err) + } + t.Skip("kernel isolation unavailable") + } +} +func startParityRuntime(t *testing.T, policy config.ArchiveContent) (*pgServeStartup, *postgres.HostedStore, *sql.DB) { + t.Helper() + requireHostedSandbox(t) + cfg, admin := hostedRuntimeConfig(t) + cfg.PG.RawDerivation = true + cfg.PG.RawPollSeconds = 1 + cfg.PG.RawMaxAttempts = 2 + cfg.ArchiveContent = policy + // An occupied archive path makes an accidental hosted archive open fail. + cfg.DBPath = filepath.Join(cfg.DataDir, "sessions.db") + require.NoError(t, os.Mkdir(cfg.DBPath, 0700)) + require.NoError(t, os.WriteFile(filepath.Join(cfg.DBPath, "sentinel"), []byte("not an archive"), 0600)) + startup, err := preparePGServeImpl(cfg, "") + require.NoError(t, err) + ctx, cancel := context.WithCancel(startup.ctx) + startup.ctx = ctx + finished := make(chan error, 1) + go func() { finished <- runPreparedPGServe(startup) }() + t.Cleanup(func() { + cancel() + select { + case err := <-finished: + assert.NoError(t, err) + case <-time.After(10 * time.Second): + t.Error("runtime failed to join") + } + startup.cleanup() + body, err := os.ReadFile(filepath.Join(cfg.DBPath, "sentinel")) + assert.NoError(t, err) + assert.Equal(t, "not an archive", string(body)) + }) + store, err := postgres.NewHostedStore(cfg.PG.URL, cfg.PG.Schema, cfg.PG.RawTenant, false) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + return &startup, store, admin +} +func newHostedCaptureClient(t *testing.T, startup *pgServeStartup, store *postgres.HostedStore, agent parser.AgentType, root string) *hostedCaptureClient { + t.Helper() + authStore, err := postgres.NewTenantRawDeviceAuthStore(store.DB(), "tenant-runtime") + require.NoError(t, err) + auth, err := rawsync.NewDeviceAuthService(authStore, time.Minute) + require.NoError(t, err) + enrolled, err := auth.EnrollDevice(t.Context(), "tenant-runtime", "synthetic capture device") + require.NoError(t, err) + provider, ok := parser.NewProvider(agent, parser.ProviderConfig{Roots: []string{root}}) + require.True(t, ok) + dir := t.TempDir() + cp, err := rawcheckpoint.OpenWithOptions(t.Context(), filepath.Join(dir, "checkpoint.db"), rawcheckpoint.Options{SpoolDir: filepath.Join(dir, "spool"), MaxOutboxBytes: 8 << 20}) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, cp.Close()) }) + require.NoError(t, cp.SetDevice(t.Context(), enrolled.Identity.DeviceID)) + c := &hostedCaptureClient{checkpoint: cp, provider: provider, device: enrolled.Identity.DeviceID, startup: startup} + reply := c.request("POST", "/api/v1/raw-sync/tokens", enrolled.Credential, "application/json", []byte(`{"scopes":["upload","commit"]}`)) + require.Equal(t, 200, reply.Code, reply.Body.String()) + var token struct{ Token string } + require.NoError(t, json.Unmarshal(reply.Body.Bytes(), &token)) + c.token = token.Token + return c +} +func (c *hostedCaptureClient) request(method, path, token, kind string, body []byte) *httptest.ResponseRecorder { + req := httptest.NewRequest(method, "http://127.0.0.1"+path, bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", kind) + req.Header.Set("X-AgentsView-Device-ID", c.device) + req.Header.Set("Upload-Offset", "0") + rec := httptest.NewRecorder() + c.startup.srv.Handler().ServeHTTP(rec, req) + return rec +} +func (c *hostedCaptureClient) commit(t *testing.T, m rawsync.Manifest) rawsync.CommitResult { + t.Helper() + body, err := json.Marshal(m) + require.NoError(t, err) + rec := c.request("POST", "/api/v1/raw-sync/manifests", c.token, "application/json", body) + require.Equal(t, 200, rec.Code, rec.Body.String()) + var result struct { + ManifestID string `json:"manifest_id"` + Receipt string `json:"receipt"` + Generation int64 `json:"generation"` + Created bool `json:"created"` + } + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &result)) + return rawsync.CommitResult{ManifestID: result.ManifestID, Receipt: result.Receipt, Generation: result.Generation, Created: result.Created} +} +func (c *hostedCaptureClient) flush(t *testing.T) rawsync.CommitResult { + t.Helper() + m, found, err := c.checkpoint.FinalizeNextManifest(t.Context(), c.device) + require.NoError(t, err) + require.True(t, found) + for _, entry := range m.Entries { + for _, ref := range entry.Objects { + body, err := json.Marshal(map[string]any{"provider": m.Provider, "object": ref}) + require.NoError(t, err) + rec := c.request("POST", "/api/v1/raw-sync/uploads", c.token, "application/json", body) + require.Contains(t, []int{200, 201}, rec.Code, rec.Body.String()) + var upload struct { + UploadID string `json:"upload_id"` + Complete bool `json:"complete"` + } + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &upload)) + if !upload.Complete { + require.NotEmpty(t, upload.UploadID) + payload, err := os.ReadFile(c.checkpoint.ObjectPath(ref)) + require.NoError(t, err) + rec = c.request("PATCH", "/api/v1/raw-sync/uploads/"+upload.UploadID, c.token, "application/octet-stream", payload) + require.Equal(t, 200, rec.Code, rec.Body.String()) + } + } + } + result := c.commit(t, m) + require.NoError(t, c.checkpoint.BindFinalizedCommit(t.Context(), c.device, m.CaptureID, result)) + _, err = c.checkpoint.AcknowledgeGeneration(t.Context(), c.device, m.CaptureID, result) + require.NoError(t, err) + c.last = m + return result +} +func (c *hostedCaptureClient) capture(t *testing.T) rawsync.CommitResult { + t.Helper() + sources, err := parser.DiscoverRawCaptureSources(t.Context(), c.provider) + require.NoError(t, err) + require.True(t, sources.Complete) + require.Len(t, sources.Sources, 1) + result, err := rawcapture.New(c.checkpoint).Capture(t.Context(), c.provider, sources.Sources[0]) + require.NoError(t, err) + require.Equal(t, rawcapture.StatusCaptured, result.Status) + return c.flush(t) +} +func waitCaptureJob(t *testing.T, pg *sql.DB, manifest, state string, attempts int) { + t.Helper() + require.Eventually(t, func() bool { + var got string + var count int + err := pg.QueryRow(`SELECT state,attempt_count FROM raw_ingest_jobs WHERE manifest_id=$1`, manifest).Scan(&got, &count) + return err == nil && got == state && count == attempts + }, 30*time.Second, 50*time.Millisecond) +} + +// Losing any captured companion, parser wire field, configured policy or +// publication child row must diverge from the independently synced SQLite rows. +func TestHostedRuntimeCapturedParity(t *testing.T) { + for _, tc := range []struct { + name string + agent parser.AgentType + policy config.ArchiveContent + }{ + {"claude", parser.AgentClaude, config.ArchiveContentFull}, {"claude_transcript", parser.AgentClaude, config.ArchiveContentTranscripts}, {"zcode", parser.AgentZCode, config.ArchiveContentFull}, {"codex_tools", parser.AgentCodex, config.ArchiveContentFull}, + } { + t.Run(tc.name, func(t *testing.T) { + startup, store, admin := startParityRuntime(t, tc.policy) + root := t.TempDir() + ids := []string{rawtest.ClaudeID} + var path string + if tc.agent == parser.AgentClaude { + path = rawtest.Claude(t, root) + } else if tc.agent == parser.AgentZCode { + rawtest.ZCode(t, root) + ids = []string{rawtest.ZCodeID, rawtest.ZCodeBillableID} + } else { + rawtest.CodexTools(t, root) + ids = []string{"codex:" + rawtest.CodexToolsID} + } + oracle, engine := rawtest.Oracle(t, tc.agent, root, tc.policy, "") + require.Equal(t, len(ids), engine.SyncAll(t.Context(), nil).Synced) + client := newHostedCaptureClient(t, startup, store, tc.agent, root) + receipt := client.capture(t) + waitCaptureJob(t, admin, receipt.ManifestID, "complete", 1) + rawtest.EqualStored(t, t.Context(), oracle, store, ids...) + core, err := postgres.NewRawProjectionStore(store.DB(), postgres.RawProjectionOptions{Tenant: "tenant-runtime"}) + require.NoError(t, err) + for _, id := range ids { + resolved, err := core.Resolve(t.Context(), id) + require.NoError(t, err) + rawtest.EqualUsageEvents(t, t.Context(), oracle, store.DB(), id, resolved.SessionID) + } + if tc.agent == parser.AgentCodex { + return + } + if tc.agent == parser.AgentZCode { + messages, err := store.GetAllMessages(t.Context(), rawtest.ZCodeBillableID) + require.NoError(t, err) + assert.Empty(t, messages) + usage, err := store.GetSessionUsage(t.Context(), rawtest.ZCodeBillableID, true) + require.NoError(t, err) + require.NotNil(t, usage) + assert.Equal(t, 17, usage.TotalOutputTokens) + assert.True(t, usage.HasCost) + return + } + if tc.policy == config.ArchiveContentFull { + require.Greater(t, len(client.last.Entries), 1, "capture must include persisted tool output") + } + require.NoError(t, store.RenameSession(rawtest.ClaudeID, new("Curated title"))) + require.NoError(t, oracle.RenameSession(rawtest.ClaudeID, new("Curated title"))) + _, err = store.StarSession(rawtest.ClaudeID) + require.NoError(t, err) + _, err = store.PinMessage(rawtest.ClaudeID, 0, new("Review this prompt")) + require.NoError(t, err) + sources, err := parser.DiscoverRawCaptureSources(t.Context(), client.provider) + require.NoError(t, err) + require.Len(t, sources.Sources, 1) + unchanged, err := rawcapture.New(client.checkpoint).Capture(t.Context(), client.provider, sources.Sources[0]) + require.NoError(t, err) + assert.Equal(t, rawcapture.StatusUnchanged, unchanged.Status) + var revision int64 + require.NoError(t, admin.QueryRow(`SELECT corpus_revision FROM raw_corpus_state`).Scan(&revision)) + replay := client.commit(t, client.last) + assert.False(t, replay.Created) + var after int64 + require.NoError(t, admin.QueryRow(`SELECT corpus_revision FROM raw_corpus_state`).Scan(&after)) + assert.Equal(t, revision, after) + waitCaptureJob(t, admin, receipt.ManifestID, "complete", 1) + rawtest.AppendClaude(t, path) + require.Equal(t, 1, engine.SyncAllForceParse(t.Context(), nil).Synced) + appended := client.capture(t) + waitCaptureJob(t, admin, appended.ManifestID, "complete", 1) + rawtest.EqualStored(t, t.Context(), oracle, store, ids...) + pins, err := store.ListPinnedMessages(t.Context(), rawtest.ClaudeID, "") + require.NoError(t, err) + require.Len(t, pins, 1) + assert.Zero(t, pins[0].Ordinal) + require.NotNil(t, pins[0].Note) + assert.Equal(t, "Review this prompt", *pins[0].Note) + stars, err := store.ListStarredSessionIDs(t.Context()) + require.NoError(t, err) + assert.Equal(t, []string{rawtest.ClaudeID}, stars) + }) + } +} + +// Equal device content coalesces, divergent content becomes explicitly +// ambiguous, and a captured tombstone removes only that device's proof. +func TestHostedRuntimeCapturedConflictAndRemoval(t *testing.T) { + startup, store, admin := startParityRuntime(t, config.ArchiveContentFull) + rootA, rootB := t.TempDir(), t.TempDir() + rawtest.Claude(t, rootA) + pathB := rawtest.Claude(t, rootB) + a := newHostedCaptureClient(t, startup, store, parser.AgentClaude, rootA) + b := newHostedCaptureClient(t, startup, store, parser.AgentClaude, rootB) + first := a.capture(t) + waitCaptureJob(t, admin, first.ManifestID, "complete", 1) + second := b.capture(t) + waitCaptureJob(t, admin, second.ManifestID, "complete", 1) + session, err := store.GetSession(t.Context(), rawtest.ClaudeID) + require.NoError(t, err) + require.NotNil(t, session) + require.NoError(t, store.RenameSession(rawtest.ClaudeID, new("Shared curation"))) + rawtest.AppendClaude(t, pathB) + divergent := b.capture(t) + waitCaptureJob(t, admin, divergent.ManifestID, "complete", 1) + _, err = store.GetSession(t.Context(), rawtest.ClaudeID) + var conflict *db.SessionIdentityError + require.ErrorAs(t, err, &conflict) + assert.Equal(t, "ambiguous", conflict.State) + require.Len(t, conflict.Variants, 2) + var contents []string + for _, id := range conflict.Variants { + messages, err := store.GetAllMessages(t.Context(), id) + require.NoError(t, err) + require.NotEmpty(t, messages) + contents = append(contents, messages[len(messages)-1].Content) + } + assert.ElementsMatch(t, []string{"The build failed.", "Recorded the failure."}, contents) + _, queued, err := b.checkpoint.QueueTombstone(t.Context(), rawcheckpoint.SourceIdentity{Provider: b.last.Provider, ConfiguredRootID: b.last.ConfiguredRootID, SourceKey: b.last.SourceKey}) + require.NoError(t, err) + require.True(t, queued) + removed := b.flush(t) + waitCaptureJob(t, admin, removed.ManifestID, "complete", 1) + session, err = store.GetSession(t.Context(), rawtest.ClaudeID) + require.NoError(t, err) + require.NotNil(t, session) + require.NotNil(t, session.DisplayName) + assert.Equal(t, "Shared curation", *session.DisplayName) + messages, err := store.GetAllMessages(t.Context(), rawtest.ClaudeID) + require.NoError(t, err) + require.NotEmpty(t, messages) + assert.Equal(t, "The build failed.", messages[len(messages)-1].Content) + _, queued, err = a.checkpoint.QueueTombstone(t.Context(), rawcheckpoint.SourceIdentity{Provider: a.last.Provider, ConfiguredRootID: a.last.ConfiguredRootID, SourceKey: a.last.SourceKey}) + require.NoError(t, err) + require.True(t, queued) + removed = a.flush(t) + waitCaptureJob(t, admin, removed.ManifestID, "complete", 1) + session, err = store.GetSession(t.Context(), rawtest.ClaudeID) + require.NoError(t, err) + assert.Nil(t, session) +} + +// A missing fork parent is a real provider partial result: publish its visible +// messages, retry finitely, and recover when capture includes the parent. +func TestHostedRuntimeCapturedPartialRetryExhaustion(t *testing.T) { + startup, store, admin := startParityRuntime(t, config.ArchiveContentFull) + root := t.TempDir() + rawtest.CodexFork(t, root) + client := newHostedCaptureClient(t, startup, store, parser.AgentCodex, root) + partial := client.capture(t) + waitCaptureJob(t, admin, partial.ManifestID, "failed", 2) + id := "codex:" + rawtest.CodexChildID + messages, err := store.GetAllMessages(t.Context(), id) + require.NoError(t, err) + require.Len(t, messages, 2) + assert.Equal(t, "Inspect the fork.", messages[0].Content) + assert.Equal(t, "Fork inspected.", messages[1].Content) + require.NoError(t, store.RenameSession(id, new("Partial curation"))) + var revision int64 + require.NoError(t, admin.QueryRow(`SELECT corpus_revision FROM raw_corpus_state`).Scan(&revision)) + replay := client.commit(t, client.last) + assert.False(t, replay.Created) + core, err := postgres.NewRawProjectionStore(store.DB(), postgres.RawProjectionOptions{Tenant: "tenant-runtime"}) + require.NoError(t, err) + rollout, err := core.ScheduleCurrentHeads(t.Context(), "same-version-replay", rawProcessingVersion(), 64) + require.NoError(t, err) + assert.True(t, rollout.Done) + // Three worker polls after failure must not claim an exhausted generation. + require.Never(t, func() bool { + var attempts int + err := admin.QueryRow(`SELECT attempt_count FROM raw_ingest_jobs WHERE manifest_id=$1`, partial.ManifestID).Scan(&attempts) + return err != nil || attempts != 2 + }, 3200*time.Millisecond, 100*time.Millisecond) + var after int64 + require.NoError(t, admin.QueryRow(`SELECT corpus_revision FROM raw_corpus_state`).Scan(&after)) + assert.Equal(t, revision, after) + rawtest.CodexParent(t, root) + sources, err := parser.DiscoverRawCaptureSources(t.Context(), client.provider) + require.NoError(t, err) + var captured bool + for _, source := range sources.Sources { + if source.Key == rawtest.CodexChildID || strings.Contains(source.DisplayPath, rawtest.CodexChildID) { + result, err := rawcapture.New(client.checkpoint).Capture(t.Context(), client.provider, source) + require.NoError(t, err) + require.Equal(t, rawcapture.StatusCaptured, result.Status) + captured = true + } + } + require.True(t, captured) + recovered := client.flush(t) + require.Greater(t, len(client.last.Entries), 1, "the parent must be captured as a companion") + waitCaptureJob(t, admin, recovered.ManifestID, "complete", 1) + oracle, engine := rawtest.Oracle(t, parser.AgentCodex, root, "", "") + require.Positive(t, engine.SyncAll(t.Context(), nil).Synced) + require.NoError(t, oracle.RenameSession(id, new("Partial curation"))) + rawtest.EqualStored(t, t.Context(), oracle, store, id) +} + +func TestHostedRuntimeCapturedRelationships(t *testing.T) { + for _, parentFirst := range []bool{true, false} { + name := "child_first" + if parentFirst { + name = "parent_first" + } + t.Run(name, func(t *testing.T) { + startup, store, admin := startParityRuntime(t, config.ArchiveContentFull) + root := t.TempDir() + rawtest.Claude(t, root) + rawtest.ClaudeChild(t, root) + oracle, engine := rawtest.Oracle(t, parser.AgentClaude, root, "", "") + require.Equal(t, 2, engine.SyncAll(t.Context(), nil).Synced) + client := newHostedCaptureClient(t, startup, store, parser.AgentClaude, root) + sources, err := parser.DiscoverRawCaptureSources(t.Context(), client.provider) + require.NoError(t, err) + require.Len(t, sources.Sources, 2) + slices.SortFunc(sources.Sources, func(a, b parser.SourceRef) int { + aChild := strings.Contains(a.DisplayPath, rawtest.ClaudeChildID) + bChild := strings.Contains(b.DisplayPath, rawtest.ClaudeChildID) + if aChild == bChild { + return 0 + } + if aChild == parentFirst { + return 1 + } + return -1 + }) + require.Equal(t, !parentFirst, strings.Contains(sources.Sources[0].DisplayPath, rawtest.ClaudeChildID)) + for _, source := range sources.Sources { + result, err := rawcapture.New(client.checkpoint).Capture(t.Context(), client.provider, source) + require.NoError(t, err) + require.Equal(t, rawcapture.StatusCaptured, result.Status) + receipt := client.flush(t) + waitCaptureJob(t, admin, receipt.ManifestID, "complete", 1) + } + rawtest.EqualStored(t, t.Context(), oracle, store, rawtest.ClaudeID, rawtest.ClaudeChildID) + children, err := store.GetChildSessions(t.Context(), rawtest.ClaudeID) + require.NoError(t, err) + require.Len(t, children, 1) + assert.Equal(t, rawtest.ClaudeChildID, children[0].ID) + + }) + } +} diff --git a/cmd/agentsview/pg_raw_sync.go b/cmd/agentsview/pg_raw_sync.go index 291873d03d..bc8146f156 100644 --- a/cmd/agentsview/pg_raw_sync.go +++ b/cmd/agentsview/pg_raw_sync.go @@ -78,6 +78,8 @@ type pgRawSyncCustody struct { mu sync.Mutex dataDir string + tenant string + objects rawsync.ObjectStore metadata rawsync.MetadataStore limits rawsync.ManifestLimits version string @@ -93,6 +95,9 @@ func (c *pgRawSyncCustody) MissingObjects( provider parser.AgentType, objects []rawsync.ObjectRef, ) ([]rawsync.ObjectRef, error) { + if c.tenant != "" && identity.TenantID != c.tenant { + return nil, rawsync.ErrUnauthorized + } service, err := c.openService(ctx) if err != nil { return nil, err @@ -107,6 +112,9 @@ func (c *pgRawSyncCustody) FinalizeObject( object rawsync.ObjectRef, body io.Reader, ) (rawsync.PutResult, error) { + if c.tenant != "" && identity.TenantID != c.tenant { + return rawsync.PutResult{}, rawsync.ErrUnauthorized + } service, err := c.openService(ctx) if err != nil { return rawsync.PutResult{}, err @@ -119,6 +127,9 @@ func (c *pgRawSyncCustody) CommitManifest( identity rawsync.AuthIdentity, manifest rawsync.Manifest, ) (rawsync.CommitResult, error) { + if c.tenant != "" && identity.TenantID != c.tenant { + return rawsync.CommitResult{}, rawsync.ErrUnauthorized + } service, err := c.openService(ctx) if err != nil { return rawsync.CommitResult{}, err @@ -153,6 +164,7 @@ func (c *pgRawSyncCustody) openService(ctx context.Context) (*rawsync.Service, e if err != nil { return fail(fmt.Errorf("preparing raw sync custody service: %w", err)) } + c.objects = objects c.repository = repository c.service = service return service, nil @@ -197,3 +209,23 @@ func combinePGRawSyncOptions(options ...server.Option) server.Option { } } } + +// The worker shares custody's single lazily opened exclusive repository. +func (c *pgRawSyncCustody) OpenManifest(ctx context.Context, identity rawsync.AuthIdentity, id string) (rawsync.ObjectInfo, rawsync.VerifiedObjectReader, error) { + if c.tenant != "" && identity.TenantID != c.tenant { + return rawsync.ObjectInfo{}, nil, rawsync.ErrUnauthorized + } + if _, err := c.openService(ctx); err != nil { + return rawsync.ObjectInfo{}, nil, err + } + return c.objects.OpenManifest(ctx, identity, id) +} +func (c *pgRawSyncCustody) CopyObject(ctx context.Context, tenant string, ref rawsync.ObjectRef, w io.Writer) (rawsync.ObjectInfo, error) { + if c.tenant != "" && tenant != c.tenant { + return rawsync.ObjectInfo{}, rawsync.ErrUnauthorized + } + if _, err := c.openService(ctx); err != nil { + return rawsync.ObjectInfo{}, err + } + return c.objects.CopyObject(ctx, tenant, ref, w) +} diff --git a/cmd/agentsview/transport.go b/cmd/agentsview/transport.go index abc45658b7..f68329caca 100644 --- a/cmd/agentsview/transport.go +++ b/cmd/agentsview/transport.go @@ -83,6 +83,20 @@ var openPGReadStore = func( pgCfg config.PGConfig, ) (db.Store, func(), error) { applyClassifierConfig(cfg) + if err := pgCfg.ValidateRawDerivation(cfg.RequireAuth); err != nil { + return nil, nil, err + } + if pgCfg.RawTenant != "" { + store, err := postgres.NewHostedStore(pgCfg.URL, pgCfg.Schema, pgCfg.RawTenant, pgCfg.AllowInsecure) + if err != nil { + return nil, nil, err + } + if err = applyRequiredCursorSecret(store, cfg); err != nil { + store.Close() + return nil, nil, err + } + return store, func() { _ = store.Close() }, nil + } store, err := postgres.NewStore( pgCfg.URL, pgCfg.Schema, pgCfg.AllowInsecure, ) diff --git a/docs/hosted-raw-sync.md b/docs/hosted-raw-sync.md index 292995accd..37d9f3969c 100644 --- a/docs/hosted-raw-sync.md +++ b/docs/hosted-raw-sync.md @@ -1,4 +1,5 @@ --- +last_edited: 2026-09-11 title: Hosted Raw Sync description: Keep original session files in hosted custody with authenticated, resumable uploads --- @@ -11,48 +12,230 @@ across restarts. `agentsview raw-sync watch` keeps the hosted copy current. ```mermaid flowchart LR Watcher["Laptop watcher"] -->|"authenticated raw upload"| Custody["Immutable raw custody"] - Custody -. "future" .-> Parser["Server parsing"] + Custody --> Parser["Isolated server parsing"] Parser --> PostgreSQL["PostgreSQL projection"] - PostgreSQL --> Embeddings["Server embeddings"] + PostgreSQL -. "future consumer" .-> Embeddings["Server embeddings"] ``` -The raw archive gives an operator the source material needed to rebuild derived -data. Version 0.42.0 ships capture and upload; it does not yet parse accepted -generations into hosted sessions or build server-owned embeddings. +The server can now parse accepted generations directly into PostgreSQL. Enable +`raw_derivation` on an explicitly provisioned hosted tenant to make uploaded +sessions browsable. Hosted processing uses no SQLite archive intermediary; +SQLite databases captured from providers remain valid source artifacts. +Embedding work is durably queued, but its consumer is not implemented. + +Device enrollment and revocation remain operator-managed. The operator supplies +each laptop with a server URL, device ID and credential. There is no public +enrollment command or HTTP endpoint. The broader delivery work remains tracked +in [issue #1352](https://github.com/kenn-io/agentsview/issues/1352), including +embedding consumption, retention, garbage collection and disaster rebuilds. + +## Provision a hosted instance + +Use one PostgreSQL schema, restricted runtime role and server instance per +tenant. Requests cannot select arbitrary tenants. The schema is permanently +bound to its tenant; runtime connections check the binding, forced row-level +security, constraints, indexes and protected catalog before serving or leasing +work. Provisioning and upgrades require a separate schema-owner connection. + +Configure an owner target and a runtime target in the operator's protected +configuration. Supply actual connection URLs and a generated cursor secret +through your secret manager or protected config file. The values below are +placeholders, not environment-variable interpolation. Use the same tenant and +schema for both targets, and make the runtime target the effective default: + +```toml +default_pg = "hosted" +require_auth = true +cursor_secret = "REPLACE_WITH_BASE64_RANDOM_SECRET" + +[pg.provision] +url = "postgres://hosted_owner@db.example.com/agentsview?sslmode=require" +schema = "hosted_sessions" +raw_tenant = "tenant-example" + +[pg.hosted] +url = "postgres://hosted_runtime@db.example.com/agentsview?sslmode=require" +schema = "hosted_sessions" +raw_tenant = "tenant-example" +raw_derivation = true +raw_poll_seconds = 5 +raw_attempt_seconds = 60 +raw_max_attempts = 5 +``` + +`cursor_secret` is a stable base64-encoded secret shared by restarts of this +instance. Keep authentication enabled even on loopback. Supply TLS through your +reverse proxy and configure the exact public origin as for ordinary remote +access. The shared server bearer token protects viewer APIs; device credentials +and scoped tokens separately protect raw-sync routes. + +Run explicit provisioning with the owner target: + +```bash +agentsview pg hosted-provision provision +``` -!!! note "You need provisioned device credentials" +This command installs or upgrades hosted tables and protections. It does not +create login roles or grant runtime access. Existing derived rows are retained; +existing raw rows must already belong to the chosen tenant. Unknown relations, +unmanaged vector layouts and conflicting ownership can block adoption. Plan +imports and schema upgrades during an operator-controlled maintenance window. +Runtime startup never migrates or provisions the hosted schema. + +Create a separate login role with `NOSUPERUSER NOBYPASSRLS NOCREATEDB +NOCREATEROLE NOREPLICATION NOINHERIT`, provision its credential through your +normal PostgreSQL administration process, and grant it `CONNECT` to the +database. It must own no application objects, have no role memberships, database +or schema `CREATE`, sibling-schema data access, or callable application +`SECURITY DEFINER` functions. Remove inherited `PUBLIC` grants where necessary, +including `CREATE` on the public schema on older PostgreSQL installations. Do +not grant `TRUNCATE`, `TRIGGER` or `REFERENCES` on hosted tables. + +For a newly provisioned schema, the following grants cover full and transcript +content, custody, publication, bounded reparse and supported curation. Replace +`hosted_sessions` and `hosted_runtime` with your schema and restricted role. Run +this as the owner before starting the runtime: - The laptop command is ready to use once the hosted deployment operator gives you - a server URL, device ID, and device credential. Device enrollment and revocation - are operator-managed; AgentsView does not yet provide a public enrollment - command or HTTP endpoint. +```sql +GRANT USAGE ON SCHEMA hosted_sessions TO hosted_runtime; +GRANT SELECT ON ALL TABLES IN SCHEMA hosted_sessions TO hosted_runtime; + +GRANT INSERT ON hosted_sessions.raw_device_tokens, + hosted_sessions.raw_manifest_entries, hosted_sessions.raw_manifest_objects, + hosted_sessions.messages, hosted_sessions.tool_calls, + hosted_sessions.tool_result_events, hosted_sessions.usage_events, + hosted_sessions.secret_findings, hosted_sessions.excluded_sessions, + hosted_sessions.raw_projection_generations, + hosted_sessions.raw_source_contributions, + hosted_sessions.raw_session_public_aliases, + hosted_sessions.raw_embedding_outbox TO hosted_runtime; + +GRANT INSERT, UPDATE ON hosted_sessions.raw_objects, + hosted_sessions.raw_source_heads, hosted_sessions.raw_ingest_jobs, + hosted_sessions.raw_source_projections, hosted_sessions.raw_session_groups, + hosted_sessions.raw_content_revisions, hosted_sessions.raw_session_branches, + hosted_sessions.raw_corpus_state, hosted_sessions.raw_projection_rollouts + TO hosted_runtime; +GRANT INSERT ON hosted_sessions.raw_manifests TO hosted_runtime; + +GRANT INSERT, UPDATE, DELETE ON hosted_sessions.raw_upload_sessions, + hosted_sessions.sessions, hosted_sessions.session_sources, + hosted_sessions.pinned_messages, hosted_sessions.raw_curation, + hosted_sessions.raw_pins TO hosted_runtime; +GRANT INSERT, DELETE ON hosted_sessions.starred_sessions, + hosted_sessions.raw_session_links TO hosted_runtime; + +GRANT USAGE ON SEQUENCE hosted_sessions.raw_ingest_jobs_id_seq, + hosted_sessions.tool_calls_id_seq, hosted_sessions.tool_result_events_id_seq, + hosted_sessions.usage_events_id_seq, hosted_sessions.pinned_messages_id_seq + TO hosted_runtime; +``` - Use [`agentsview pg push`](/docs/pg-sync/) when the shared server must provide - browsable sessions today. It parses sessions locally and can build embeddings - locally before pushing derived rows and vectors to PostgreSQL. +Sequence names above are those created by provisioning. For an adopted schema, +resolve the actual owned sequence with `pg_get_serial_sequence`; serial IDs need +`USAGE` or `UPDATE`, while identity-generated IDs need no separate sequence +grant. Enrollment needs an operator credential with device-write privileges; the +runtime grants intentionally omit those privileges. -The tracked delivery sequence and production acceptance criteria live in -[GitHub issue #1352](https://github.com/kenn-io/agentsview/issues/1352). +`archive_content = "usage"` also requires `SELECT` on `vector_generations` and +`SELECT, DELETE` on existing `vector_documents`, `vector_push_state` and each +existing `vector_chunks_g` table named by a generation. This removes any +previously retained indexed content. It does not create or consume embeddings. +The runtime checks the configured policy's grants before readiness and reports +missing privileges instead of silently disabling hosted processing. -## Delivery status +Start the effective default target: + +```bash +agentsview pg serve --no-browser +``` + +Named targets retain their ordinary selection rules. A single-target deployment +can put the same hosted keys under `[pg]`. `AGENTSVIEW_PG_URL` and +`AGENTSVIEW_PG_SCHEMA` override only the effective default target. They do not +rewrite the separate named owner target. `pg push` refuses a hosted-owned schema +before mutation, even if the client omits its hosted configuration fields. Use a +separate legacy schema for local pushes. + +## Isolation and processing limits + +Hosted parser activation requires Linux amd64 or arm64, a cgo-enabled build, +user/mount/network namespaces, `close_range`, and seccomp with thread +synchronization. Startup tests actual source visibility inside the sandbox +before claiming a job. Unsupported kernels, containers, non-Linux hosts and +Linux builds without cgo fail closed. There is no in-process parser fallback. +The positive kernel suite has been executed on amd64; arm64 has compile proof +but still needs an execution gate on that architecture. + +Each parser child gets only bounded protocol pipes and a minimal environment. A +pre-runtime constructor closes inherited descriptors above stderr before Go +initializes. The child sees a read-only source mount inside an otherwise empty, +read-only jail. Filesystem escape, networking and process creation are denied; +seccomp applies to every existing thread and allows only constrained runtime +thread creation. No external sandbox helper is required. + +| Limit | Value | +| -------------------------------------------- | ----------------------------------------------- | +| Worker concurrency | One sequential job per instance | +| Poll interval | Default 5 seconds; maximum 60 | +| Whole attempt wall time | Default 60 seconds; maximum 300 | +| Attempts per selected generation | Default 5; maximum 10 | +| Retry backoff | Exponential from 1 second, capped at 60 seconds | +| Lease / heartbeat | 60 seconds / 10 seconds | +| Materialized source bytes | 512 MiB | +| Parser stdout / stderr | 32 MiB / 64 KiB | +| Child address space / data | 2 GiB / 512 MiB | +| Child CPU / open descriptors | 30 seconds / 64 | +| Manifest bytes / entries / object references | 1 MiB / 4,096 / 16,384 | + +Custody may accept files larger than the materialization limit. Such captures +remain retained but cannot be projected by this worker. Accepted source bytes +are untrusted, even after device authentication. Publication atomically writes +normalized rows, source proof, identity changes, curation and job outcome under +the selected generation and lease fence. Partial results retain older proof for +unresolved members and retry finitely. Exhausted jobs remain failed until new +source or processing-version selection provides new work. + +Equivalent content from several devices coalesces when it shares a stable +provider session identity. Without that identity, sources remain separate even +when their content matches. Divergent content makes the +bare session ID ambiguous and exposes explicit variants. Source removal retracts +only that source's proof. Names, stars and pins survive compatible publication; +ambiguous identity never silently picks a transcript. Owner imports that change +legacy identity must retry their transaction on serialization failure (`SQLSTATE +40001`) if publication or curation holds a conflicting identity lock. Ordinary +legacy content and curation writes do not take those identity locks. + +The maintenance pass examines at most 64 indexed pending signal rows with a +10-second timeout. It does not scan the archive during idle polling. Shutdown +cancels and joins worker, materializer and parser work before closing custody +and PostgreSQL. + +## Reparse and rollback + +After an executable upgrade changes the parser data version, schedule current +heads explicitly in bounded batches: + +```bash +agentsview pg raw-reparse hosted --run-id parser-rollout-1 --batch-size 64 +``` -| Layer | Status | Current boundary | -| ---------------------- | ------------- | ---------------------------------------------------------------------------------------------------------------------------- | -| Raw custody | Available | Validated objects, canonical manifests, durable receipts, source-head fencing, and parse-job creation | -| Device authentication | Available | Credential exchange, scoped short-lived tokens, server-derived identity, and revocation; enrollment remains operator-managed | -| HTTP raw transport | Available | Missing-object negotiation, resumable upload, and manifest commit; status is local only | -| Laptop capture | Available | Watching, bounded audits, safe SQLite snapshots, durable spooling, checkpoints, retries, and local status | -| Server derivation | Not available | Accepted generations are not yet parsed into PostgreSQL sessions or embeddings | -| Operations and cutover | Not available | Retention, garbage collection, disaster rebuilds, and migration from `pg push` remain future work | +Repeat the same run ID until the command output contains `complete=true`. A call +selects at most 1–256 heads and atomically saves its keyset checkpoint. The run +ID is bound to the executable's processing version. Equal manifest/version +selection is idempotent: a new run ID does not resurrect completed or exhausted +jobs for that same selection. Startup and idle polls perform no reparse scan. -The server parse-worker foundation now includes fenced PostgreSQL job leases, -verified source materialization, provider parsing, retry handling, and a -projection interface. It is an internal library: `pg serve` does not start a -worker, and a PostgreSQL session-projection implementation is still pending. -Hosted browsing and embeddings therefore continue to require `pg push`. +To stop derivation, set `raw_derivation = false` and restart. Keep `raw_tenant`, +authentication, the cursor secret and the tenant-bound runtime connection. +Hosted public reads and raw custody remain available; accepted new manifests +still select their current processing version. Removing `raw_tenant` from an +owned schema fails rather than exposing physical storage identities. This +rollback does not convert the schema back to a `pg push` destination. -The broader delivery issue remains open because public enrollment, hosted -session derivation, and production lifecycle controls are not finished. +Keep PostgreSQL metadata and the immutable raw repository together in backups. +Automated retention, garbage collection, disaster rebuilds, enrollment UX, +embedding consumption and migration cutover tooling remain outside this release. ## Laptop raw watch daemon @@ -82,10 +265,11 @@ doing so intentionally creates two watchers over the same provider roots. ## HTTP control plane -`agentsview pg serve` registers the raw-sync routes when its PostgreSQL role can -write every raw-sync table and the ingest-job sequence. A read-only role keeps -serving the normal PostgreSQL-backed UI and API without these runtime routes. -There is no separate raw-sync configuration switch. When requirements are +In legacy mode, `agentsview pg serve` registers raw-sync routes when its +PostgreSQL role has the required custody-table and ingest-job sequence grants. A +read-only role keeps serving the PostgreSQL UI and API without these routes. +Explicit hosted mode uses `raw_tenant` and requires the full hosted preflight; +`raw_derivation` controls its worker. When legacy route requirements are missing, startup logs `raw-sync routes disabled; missing requirements:` followed by the exact missing table privileges, sequence access, or read-only transaction setting. @@ -191,10 +375,20 @@ The implemented foundations isolate object and metadata identities by tenant and do not deduplicate across tenants. A production deployment must also provide TLS in transit, encryption at rest for object storage, PostgreSQL, backups, and worker scratch space, plus access controls around device enrollment and -revocation. PostgreSQL row-level security remains a planned defense-in-depth -layer; the current foundation does not configure it. +revocation. Explicit hosted provisioning installs forced PostgreSQL row-level +security, tenant constraints and schema binding; runtime validation rejects +weakened protections. This is one tenant per instance, not request-multiplexed +tenancy. Treat the HTTP routes as the protocol between the bundled laptop client and a -hosted AgentsView deployment, not as a general integration API. Public operator -controls, compatibility policy, and recovery tooling will be documented when -those entry points exist. +hosted AgentsView deployment, not as a general integration API. Enrollment and +lifecycle controls remain operator-managed; the provisioning and bounded reparse +commands above are the implemented operator entry points. + +Explicit parent and tool-subagent links first use their own source's historical +proof. Removed or excluded same-source proof prevents fallback. When no such +proof exists, a link may resolve across files only within the same tenant, +device, provider, and configured root, and only to one eligible content cohort. +Conflicting candidates remain unresolved. Fresh session, timing, and sidebar +reads use this rule. A target-only graph change does not necessarily notify an +unchanged owner's session stream; graph-only live refresh is not guaranteed. diff --git a/docs/internal/session-format-sources.md b/docs/internal/session-format-sources.md index 59f9cb63d8..c2d0f32340 100644 --- a/docs/internal/session-format-sources.md +++ b/docs/internal/session-format-sources.md @@ -1,3 +1,7 @@ +--- +last_edited: 2026-09-11 +--- + # Session Format Source Inventory This inventory records the best reproducible evidence currently available for @@ -101,6 +105,11 @@ add an archived or maintained mirror without replacing the original identity. ## Claude Code (`claude`) +Rechecked 2026-09-11 against the existing provider parser and its metadata +fixtures: the first nonempty JSONL `sessionId` supplies `SourceSessionID`. A +filename alone does not supply that provider identity. Hosted multi-device +fixtures retain this field; missing identities remain source-local. + - **Performance fixture check (2026-09-04):** Rechecked the pinned Codeburn format notes below for project-scoped JSONL. `cmd/perfsim` uses the shared Claude fixture builder to emit user/assistant pairs with message/request @@ -268,7 +277,12 @@ add an archived or maintained mirror without replacing the original identity. background flag leave lineage unresolved when their complete UUID sets differ. Identical sets elect the smallest stem, retaining one copy across three background transcripts. An interactive original still wins a tie with - a background sibling. Reverified 2026-08-16 with Claude Code 2.1.233 using a + a background sibling. Reverified 2026-09-11 with same-size sibling rewrites + that restore mtime: filesystem ctime can also remain unchanged. Cached head + metadata therefore requires a bounded leading-byte digest check before + reusing the root UUID or background stamp. Full parsing retains the verified + parent and trims only the replayed prefix; no provider format changed. + Reverified 2026-08-16 with Claude Code 2.1.233 using a controlled `claude -p --session-id ` probe under an isolated `CLAUDE_CONFIG_DIR`. Before the deliberately bounded probe was terminated during its API retry, Claude had created the exact UUID transcript under @@ -497,6 +511,12 @@ add an archived or maintained mirror without replacing the original identity. orphaned child's full transcript when its named parent is unavailable, matching local parsing. When available, the explicitly named parent travels with the captured fork so hosted parsing applies the local replay boundary. + Rechecked 2026-09-11 against the pinned protocol's `forked_from_id` field + and synthetic registered-provider fixtures: an absent parent reports + `DataVersionNeedsRetry` while retaining the child messages; a later captured + readable turnless parent resolves the same child as current. The hosted + integration exercises finite retry exhaustion and new-generation recovery + without changing these provider semantics. Reverified on 2026-09-10 with `TestProviderParserHostedParseMatchesLocalCodexForkLineage`: parents in other configured homes, archives, and custom roots also travel with the @@ -735,7 +755,13 @@ add an archived or maintained mirror without replacing the original identity. available; monetary cost is catalog-derived. - **Agentsview:** `internal/parser/gemini.go` and `internal/parser/gemini_provider.go`; both JSON and JSONL generations remain - supported. + supported. Reverified the pinned recording source on 2026-09-11: token + metadata is attached separately from message content; its writer emits zero + defaults when usage arrives. Hosted transport preserves the parser’s explicit + coverage state, including absent usage and partially populated older records, + rather than inferring coverage from normalized zero-valued keys. The + provider-wire-preparation regression is + `TestSandboxGeminiWirePreservesTokenCoverage`. ## Gemini Apps (`gemini-apps`) diff --git a/docs/pg-sync.md b/docs/pg-sync.md index dc8641d8cb..349c96e964 100644 --- a/docs/pg-sync.md +++ b/docs/pg-sync.md @@ -1,4 +1,5 @@ --- +last_edited: 2026-09-11 title: PostgreSQL Sync description: Share sessions across machines with PostgreSQL push sync, an auto-push service, and a read-only server --- @@ -21,9 +22,28 @@ dashboard as well. [Hosted Raw Sync](/docs/hosted-raw-sync/) is a separate path that keeps original provider artifacts in hosted custody through authenticated, resumable uploads. - It does not yet turn accepted generations into hosted sessions or embeddings. - This does not change `pg push`, `pg push --watch`, or the read-only session UI - and APIs documented on this page. + Explicitly provisioned tenants can enable `raw_derivation` to parse accepted + sources directly into browsable PostgreSQL sessions, without a local SQLite + archive. Embedding consumption remains future work. Ordinary `pg push` and + read-only PostgreSQL serving keep their existing behavior; `pg push` refuses + a schema that has been adopted for hosted raw processing. + +## Hosted raw processing + +For uploads that the server parses itself, provision a schema and tenant with +`agentsview pg hosted-provision `, then serve with a separate +restricted runtime role. The selected PG target needs `raw_tenant`, an explicit +schema and `raw_derivation = true`; the server also requires authentication, a +stable cursor secret and a supported Linux/cgo sandbox. Runtime startup checks +schema protections and grants but never runs hosted migrations. + +See [Hosted Raw Sync](/docs/hosted-raw-sync/#provision-a-hosted-instance) for +the configuration, least-privilege grants and platform limits. Use `agentsview +pg raw-reparse --run-id --batch-size 64` for bounded +parser-version rollouts. To stop the worker, set `raw_derivation = false` and +keep `raw_tenant`, authentication and the cursor secret. This preserves hosted +public reads and raw custody. It does not restore `pg push` access to the owned +schema. ## Quick Start @@ -45,7 +65,9 @@ For multiple PostgreSQL destinations, use named `[pg.NAME]` blocks and `default_pg` instead of the legacy single `[pg]` block. Named target names are normalized case-insensitively, and `all`, `local`, plus the legacy `[pg]` field names `url`, `schema`, `machine_name`, `allow_insecure`, `projects`, and -`exclude_projects` are unavailable as `[pg.NAME]` names. +`exclude_projects`, `raw_tenant`, `raw_derivation`, `raw_poll_seconds`, +`raw_attempt_seconds`, and `raw_max_attempts` are unavailable as `[pg.NAME]` +names. ### 2. Push Sessions @@ -317,7 +339,8 @@ to run it. ### `agentsview pg serve` -Start a read-only web UI backed by PostgreSQL. +Start the PostgreSQL web UI. Legacy mode reads pushed sessions; an explicitly +provisioned hosted target can also derive sessions from accepted raw captures. ```bash agentsview pg serve [flags] @@ -358,14 +381,14 @@ GRANT SELECT, UPDATE ON agentsview.raw_ingest_jobs TO raw_sync_runtime; Run this as the schema owner, substitute your schema and runtime role, and restart `pg serve`. Keep the existing `INSERT` and ingest-job sequence `USAGE` -grants. This enables the raw-sync HTTP routes; `pg serve` does not yet start the -internal parse worker or project hosted raw captures into browsable sessions. +grants. This enables legacy raw-custody HTTP routes. For server-owned parsing, +use the explicit hosted setup below. -On startup, `pg serve` automatically applies any pending schema migrations to -PostgreSQL, creating new tables and indexes added in newer AgentsView versions. -This removes the need to run `pg push` before starting the server after an -upgrade. If the PostgreSQL role is read-only, the migration is skipped and the -server falls back to the schema compatibility check. +In legacy mode, `pg serve` automatically applies any pending schema migrations +to PostgreSQL, creating new tables and indexes added in newer AgentsView +versions. This removes the need to run `pg push` before starting the server +after an upgrade. If the PostgreSQL role is read-only, the migration is skipped +and the server falls back to the schema compatibility check. When `require_auth` is enabled, a bearer token is generated if needed and printed on startup. Pass it via `Authorization: Bearer ` on API requests. @@ -586,7 +609,7 @@ ______________________________________________________________________ are not deleted from PostgreSQL because the local rows no longer exist at push time. Use a direct SQL DELETE to clean up PostgreSQL if needed. Soft-deleted sessions (trash) sync correctly. -- **Schema compatibility** — `pg serve` automatically applies pending schema +- **Schema compatibility** — legacy `pg serve` automatically applies pending schema migrations on startup. If the PostgreSQL role lacks DDL permissions, run `agentsview pg push` from a machine with write access to update the schema. - **Trigram index bloat on pre-0.33.0 schemas** — the content search index was diff --git a/frontend/messages/en.json b/frontend/messages/en.json index 71c3ee6a41..358d212665 100644 --- a/frontend/messages/en.json +++ b/frontend/messages/en.json @@ -1726,6 +1726,7 @@ "pinned_expand": "Expand", "pinned_copy_message": "Copy message", "pinned_copied_message": "Copied message", + "pinned_unresolved": "Pinned message is no longer in this transcript.", "pinned_unpin": "Unpin", "settings_session_providers_title": "Session Providers", "settings_session_providers_description": "Choose which agent session sources AgentsView discovers and syncs.", diff --git a/frontend/messages/fr.json b/frontend/messages/fr.json index 99a955880a..1e71ebaa9a 100644 --- a/frontend/messages/fr.json +++ b/frontend/messages/fr.json @@ -1725,6 +1725,7 @@ "pinned_expand": "Déplier", "pinned_copy_message": "Copier le message", "pinned_copied_message": "Message copié", + "pinned_unresolved": "Le message épinglé ne figure plus dans cette conversation.", "pinned_unpin": "Désépingler", "settings_session_providers_title": "Fournisseurs de sessions", "settings_session_providers_description": "Choisissez les sources de sessions d’agents qu’AgentsView détecte et synchronise.", diff --git a/frontend/messages/ja.json b/frontend/messages/ja.json index ddd02bba9b..802f35078c 100644 --- a/frontend/messages/ja.json +++ b/frontend/messages/ja.json @@ -1726,6 +1726,7 @@ "pinned_expand": "展開する", "pinned_copy_message": "メッセージをコピーする", "pinned_copied_message": "コピーされたメッセージ", + "pinned_unresolved": "ピン留めしたメッセージはこの会話に存在しません。", "pinned_unpin": "ピン留めを外す", "settings_session_providers_title": "セッションプロバイダー", "settings_session_providers_description": "AgentsView が検出して同期するエージェント セッション ソースを選択します。", diff --git a/frontend/messages/ko.json b/frontend/messages/ko.json index cd8cf0f040..f6a8a29891 100644 --- a/frontend/messages/ko.json +++ b/frontend/messages/ko.json @@ -1664,6 +1664,7 @@ "pinned_expand": "펼치기", "pinned_copy_message": "메시지 복사", "pinned_copied_message": "메시지가 복사됨", + "pinned_unresolved": "고정한 메시지가 이 대화에 더 이상 없습니다.", "pinned_unpin": "고정 해제", "settings_session_providers_title": "세션 공급자", "settings_session_providers_description": "AgentsView가 검색하고 동기화할 에이전트 세션 소스를 선택합니다.", diff --git a/frontend/messages/zh-CN.json b/frontend/messages/zh-CN.json index cbc6e983f6..1d99efcdc4 100644 --- a/frontend/messages/zh-CN.json +++ b/frontend/messages/zh-CN.json @@ -1662,6 +1662,7 @@ "pinned_expand": "展开", "pinned_copy_message": "复制消息", "pinned_copied_message": "已复制消息", + "pinned_unresolved": "已固定的消息已不在此会话中。", "pinned_unpin": "取消固定", "settings_session_providers_title": "会话提供商", "settings_session_providers_description": "选择 AgentsView 要发现和同步的代理会话来源。", diff --git a/frontend/messages/zh-TW.json b/frontend/messages/zh-TW.json index 0663332de2..1e4bf59dca 100644 --- a/frontend/messages/zh-TW.json +++ b/frontend/messages/zh-TW.json @@ -1662,6 +1662,7 @@ "pinned_expand": "展開", "pinned_copy_message": "複製訊息", "pinned_copied_message": "已複製訊息", + "pinned_unresolved": "已釘選的訊息已不在此對話中。", "pinned_unpin": "取消釘選", "settings_session_providers_title": "工作階段提供者", "settings_session_providers_description": "選擇 AgentsView 要探索及同步的代理工作階段來源。", diff --git a/frontend/src/lib/api/generated/models/apiErrorResponse.ts b/frontend/src/lib/api/generated/models/apiErrorResponse.ts index 5779ad44ce..4862a300f0 100644 --- a/frontend/src/lib/api/generated/models/apiErrorResponse.ts +++ b/frontend/src/lib/api/generated/models/apiErrorResponse.ts @@ -8,5 +8,7 @@ export interface ApiErrorResponse { current_manifest_id?: string; current_receipt?: string; error: string; + state?: string; upload_offset?: number; + variants?: string[]; } diff --git a/frontend/src/lib/api/generated/models/configPGConfig.ts b/frontend/src/lib/api/generated/models/configPGConfig.ts index cec70bb44c..6a10dfff52 100644 --- a/frontend/src/lib/api/generated/models/configPGConfig.ts +++ b/frontend/src/lib/api/generated/models/configPGConfig.ts @@ -8,6 +8,11 @@ export interface ConfigPGConfig { machine_name: string; projects?: string[]; push_vectors?: boolean; + raw_attempt_seconds?: number; + raw_derivation?: boolean; + raw_max_attempts?: number; + raw_poll_seconds?: number; + raw_tenant?: string; schema: string; url: string; } diff --git a/frontend/src/lib/api/generated/models/dbPinnedMessage.ts b/frontend/src/lib/api/generated/models/dbPinnedMessage.ts index 104cd58ce0..f08bc83f50 100644 --- a/frontend/src/lib/api/generated/models/dbPinnedMessage.ts +++ b/frontend/src/lib/api/generated/models/dbPinnedMessage.ts @@ -7,6 +7,7 @@ export interface DbPinnedMessage { created_at: string; id: number; message_id: number; + message_key?: string; note?: string; ordinal: number; role?: string; @@ -15,4 +16,5 @@ export interface DbPinnedMessage { session_first_message?: string; session_id: string; session_project?: string; + unresolved?: boolean; } diff --git a/frontend/src/lib/api/generated/models/dbSession.ts b/frontend/src/lib/api/generated/models/dbSession.ts index 37071b614a..e7f1c3dc24 100644 --- a/frontend/src/lib/api/generated/models/dbSession.ts +++ b/frontend/src/lib/api/generated/models/dbSession.ts @@ -42,6 +42,7 @@ export interface DbSession { outcome: string; outcome_confidence: string; parent_session_id?: string; + parent_session_ids?: string[]; parser_malformed_lines?: number; peak_context_tokens: number; project: string; diff --git a/frontend/src/lib/api/generated/models/dbSidebarSessionIndexRow.ts b/frontend/src/lib/api/generated/models/dbSidebarSessionIndexRow.ts index 7ed28bac72..911e179345 100644 --- a/frontend/src/lib/api/generated/models/dbSidebarSessionIndexRow.ts +++ b/frontend/src/lib/api/generated/models/dbSidebarSessionIndexRow.ts @@ -16,6 +16,7 @@ export interface DbSidebarSessionIndexRow { machine: string; message_count: number; parent_session_id?: string; + parent_session_ids?: string[]; project: string; relationship_type?: string; session_kind?: string; diff --git a/frontend/src/lib/api/generated/models/deleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters.ts b/frontend/src/lib/api/generated/models/deleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters.ts new file mode 100644 index 0000000000..89b1390984 --- /dev/null +++ b/frontend/src/lib/api/generated/models/deleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters.ts @@ -0,0 +1,8 @@ +/** + * Generated by Orval. Do not edit manually. + */ + +export type DeleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters = { + id: string; + messageKey: string; +}; diff --git a/frontend/src/lib/api/generated/models/index.ts b/frontend/src/lib/api/generated/models/index.ts index 534b1be1f4..0d06f97982 100644 --- a/frontend/src/lib/api/generated/models/index.ts +++ b/frontend/src/lib/api/generated/models/index.ts @@ -166,6 +166,7 @@ export * from "./deleteApiV1InsightsByIdPathParameters.ts"; export * from "./deleteApiV1SessionsByIdMessagesByMessageIdPinPathParameters.ts"; export * from "./deleteApiV1SessionsByIdPathParameters.ts"; export * from "./deleteApiV1SessionsByIdPermanentPathParameters.ts"; +export * from "./deleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters.ts"; export * from "./deleteApiV1SessionsByIdStarPathParameters.ts"; export * from "./deleteApiV1SettingsWorktreeMappingsByIdPathParameters.ts"; export * from "./embeddingsBuildRequest.ts"; diff --git a/frontend/src/lib/api/generated/models/serviceSessionDetail.ts b/frontend/src/lib/api/generated/models/serviceSessionDetail.ts index 9dc7e9f8aa..7b213cdd1d 100644 --- a/frontend/src/lib/api/generated/models/serviceSessionDetail.ts +++ b/frontend/src/lib/api/generated/models/serviceSessionDetail.ts @@ -46,6 +46,7 @@ export interface ServiceSessionDetail { outcome: string; outcome_confidence: string; parent_session_id?: string; + parent_session_ids?: string[]; parser_malformed_lines?: number; peak_context_tokens: number; project: string; diff --git a/frontend/src/lib/api/generated/pins/pins.ts b/frontend/src/lib/api/generated/pins/pins.ts index f40bd65bd2..b6547ad1e0 100644 --- a/frontend/src/lib/api/generated/pins/pins.ts +++ b/frontend/src/lib/api/generated/pins/pins.ts @@ -3,6 +3,7 @@ */ import type { DeleteApiV1SessionsByIdMessagesByMessageIdPinPathParameters, + DeleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters, GetApiV1PinsParams, GetApiV1SessionsByIdPinsPathParameters, PinMessageResponse, @@ -94,6 +95,29 @@ export const postApiV1SessionsByIdMessagesByMessageIdPin = async ( ); }; +export const getDeleteApiV1SessionsByIdPinReferencesByMessageKeyUrl = ({ + id, + messageKey, +}: DeleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters) => { + return `/api/v1/sessions/${encodeURIComponent(String(id))}/pin-references/${encodeURIComponent(String(messageKey))}`; +}; + +/** + * @summary Remove retained pin + */ +export const deleteApiV1SessionsByIdPinReferencesByMessageKey = async ( + { id, messageKey }: DeleteApiV1SessionsByIdPinReferencesByMessageKeyPathParameters, + options?: Parameters[1], +): Promise => { + return orvalFetch( + getDeleteApiV1SessionsByIdPinReferencesByMessageKeyUrl({ id, messageKey }), + { + ...options, + method: "DELETE", + }, + ); +}; + export const getGetApiV1SessionsByIdPinsUrl = ({ id }: GetApiV1SessionsByIdPinsPathParameters) => { return `/api/v1/sessions/${encodeURIComponent(String(id))}/pins`; }; diff --git a/frontend/src/lib/api/types/core.ts b/frontend/src/lib/api/types/core.ts index 99f944d03d..43a5277432 100644 --- a/frontend/src/lib/api/types/core.ts +++ b/frontend/src/lib/api/types/core.ts @@ -36,6 +36,7 @@ export interface Session { message_count: number; user_message_count: number; parent_session_id?: string; + parent_session_ids?: string[]; relationship_type?: string; deleted_at?: string | null; termination_status?: string | null; @@ -92,6 +93,7 @@ export interface SessionPage { export interface SidebarSessionIndexRow { id: string; parent_session_id?: string | null; + parent_session_ids?: string[]; relationship_type?: string | null; project: string; machine: string; @@ -197,6 +199,8 @@ export interface AgentsResponse { /** Matches Go PinnedMessage struct in internal/db/pins.go */ export interface PinnedMessage { + message_key?: string; + unresolved?: boolean; id: number; session_id: string; message_id: number; diff --git a/frontend/src/lib/components/pinned/PinnedPage.svelte b/frontend/src/lib/components/pinned/PinnedPage.svelte index dcdc356948..7af46fc9e1 100644 --- a/frontend/src/lib/components/pinned/PinnedPage.svelte +++ b/frontend/src/lib/components/pinned/PinnedPage.svelte @@ -116,7 +116,7 @@ {:else}
- {#each pins.pins as pin (pin.id)} + {#each pins.pins as pin (`${pin.session_id}:${pin.message_key ?? pin.id}`)} {@const info = getSessionInfo(pin)} {@const isExpanded = expanded.has(pin.id)} {@const preview = previewContent(pin.content)} @@ -136,6 +136,9 @@ {formatRelativeTime(pin.created_at)}
+ {#if pin.unresolved} +
{m.pinned_unresolved()}
+ {/if} {#if preview}
{#if isExpanded && pin.content} @@ -154,6 +157,7 @@