diff --git a/docs/api-compatibility.md b/docs/api-compatibility.md index a45eeced..fefbf854 100644 --- a/docs/api-compatibility.md +++ b/docs/api-compatibility.md @@ -1,7 +1,7 @@ # REST API contract compatibility Tapper and Tapper Hub select a REST contract independently of their release -numbers. The initial and currently sole supported revision is `2026-09-11`. +numbers. The sole supported revision is `2026-09-28`, which replaced `2026-09-11` when grep, listing, and relationship responses gained server-side projection. Change the revision for incompatible API behavior. Compatible additions and fixes retain the revision; release numbers remain diagnostic metadata. @@ -9,11 +9,11 @@ fixes retain the revision; release numbers remain diagnostic metadata. returns `Cache-Control: no-store` with the actual server build version: ```json -{"server_version":"0.24.0","api_versions":["2026-09-11"]} +{"server_version":"0.24.0","api_versions":["2026-09-28"]} ``` Every request under `/api/v1`, including event-stream handshakes and telemetry, -must send exactly one `Tapper-API-Version: 2026-09-11` header. Accepted requests +must send exactly one `Tapper-API-Version: 2026-09-28` header. Accepted requests echo it in the response. Clients send their release in `Tapper-Client-Version`. The Hub validates the contract before authentication, flight authority, or application handlers. Discovery, health, OAuth, browser routes, and hosted @@ -26,7 +26,7 @@ duplicate, comma-joined, or unsupported values return HTTP 400 with `operationPerformed: false`, for example: ```json -{"error":"unsupported REST contract; upgrade Tapper and Hub together","code":"API_VERSION_UNSUPPORTED","requested_api_versions":["2099-01-01"],"supported_api_versions":["2026-09-11"],"server_version":"0.24.0","operationPerformed":false} +{"error":"unsupported REST contract; upgrade Tapper and Hub together","code":"API_VERSION_UNSUPPORTED","requested_api_versions":["2099-01-01"],"supported_api_versions":["2026-09-28"],"server_version":"0.24.0","operationPerformed":false} ``` CLI invocations and MCP connections discover each selected Hub once, before @@ -57,7 +57,7 @@ working against the new Hub, and new clients reject legacy Hubs. Update custom scripts explicitly, for example: ```sh -curl -H 'Tapper-API-Version: 2026-09-11' \ +curl -H 'Tapper-API-Version: 2026-09-28' \ -H "Authorization: Bearer ${TAPPER_TOKEN}" \ https://hub.example.com/api/v1/whoami ``` diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 29812b36..02063a53 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -49,6 +49,8 @@ touching). Two implementations exist: and maintains derived state itself. - `*keg.RemoteKeg` (`keg_remote.go`) speaks the hub's operation-level HTTP API — one request per operation; all orchestration happens server-side. + Complete TAP operations (links, backlinks, tags) forward to the Hub's + `/api/v1/tap/*` routes through `Tap.Operations`. `pkg/tapper` resolves a keg via `keg.NewKegFromTarget`, which returns the interface; `pkg/tapper` never touches `Repository` directly. All node diff --git a/docs/architecture/runtime-and-concurrency.md b/docs/architecture/runtime-and-concurrency.md index 829dcf65..781307f2 100644 --- a/docs/architecture/runtime-and-concurrency.md +++ b/docs/architecture/runtime-and-concurrency.md @@ -90,6 +90,12 @@ use of `time.Now()` that cannot be driven by a fake clock: - **Remote operations are single-request**: each `RemoteKeg` method is one HTTP round trip, and the hub serializes per-node writes server-side. There is no client-side lock lease or dex write over HTTP. +- **Orientation reuse in `tap mcp`**: each tool call needs the session's + resolved orientation, which costs several Hub requests to build. `tap mcp` + reuses it for up to ten seconds (`ServerOptions.OrientationCacheTTL`); + `orient`, `session_refresh`, flight mutations, and any failed call discard + it. Reuse cannot widen access, because the Hub rechecks the orientation + header and KEG authority on every request. Hosted servers leave it off. - **Advisory locks are session primitives**: `Keg.Lock`/`Unlock`/ `LockStatus`/`ForceUnlock` (used by `tap lock` / `tap node edit`) are opt-in advisory locks backed by the Hub's `/nodes/{id}/lock` endpoints. Leases carry a TTL diff --git a/docs/architecture/service-layer.md b/docs/architecture/service-layer.md index 630598f4..4ecf18aa 100644 --- a/docs/architecture/service-layer.md +++ b/docs/architecture/service-layer.md @@ -55,6 +55,18 @@ editor open/save, schema creation, dex reads, create, and lock acquisition therefore have matching local and hosted semantics without client fan-out. +Listings resolve the fields their format renders where the data lives: +`grep` sends its paging and field selectors with the scan, and any known +result set resolves fields with one `ListView` call restricted by `node_ids`. +No listing reads metadata one node at a time. + +Complete TAP operations go through `Tap.Operations`. `links`, `backlinks`, and +`tags` are each one unit — the lookup, paging, and the fields the format +renders — so `DefaultOperations` forwards a Hub KEG's operation to the Hub's +`/api/v1/tap/*` route (`pkg/tapapi`) in one request and composes KEG +operations only for a KEG without a Hub. A Hub serving its own MCP endpoint +injects its in-process TAP service here. + The Tap layer groups mixed-keg read and validation arguments by resolved keg, issues one batch per group, and restores caller order. Interactive editing and watching remain separate phases. Cross-keg import uses one source export, one diff --git a/docs/output-formats.md b/docs/output-formats.md index 39ac0479..59415f9a 100644 --- a/docs/output-formats.md +++ b/docs/output-formats.md @@ -36,9 +36,9 @@ that appears in both means the same thing in both. | `%{id}` | the node id | free | | `%{title}` | the node title | free | | `%{.updated}`, `%{.created}`, `%{.accessed}` | index timestamps, RFC3339 | free | -| `%{.hash}`, `%{.lead}`, `%{.accessCount}`, `%{.omega}` | statistics fields | one read per node | -| `%{tags}` | the node's tags, comma separated | one read per node | -| `%{anything-else}` | that metadata key | one read per node | +| `%{.hash}`, `%{.lead}`, `%{.accessCount}`, `%{.omega}` | statistics fields | one batch per listing | +| `%{tags}` | the node's tags, comma separated | one batch per listing | +| `%{anything-else}` | that metadata key | one batch per listing | `id`, `title`, and `tags` are reserved. A node carrying metadata under one of those keys cannot address it in field position; the intrinsic wins. @@ -127,9 +127,12 @@ Formats naming only `id`, `title`, the three dates, or the legacy verbs — whic includes the default — read nothing beyond the index that a listing already loads. -Any other selector reads one file per node, for the nodes in the result window -only. Combine with `--limit` on a large keg. The same is true of the equivalent -query predicates: filtering on `entity=plan` or `.hash=` also reads per node. +Any other selector is resolved by the Hub for the nodes in the result window +only, in one batch with the listing itself: `list` and `grep` stay a single +request, and `links`, `backlinks`, and a `tags` query add one listing request +however many rows they return. Paging (`--limit`, `--offset`) happens before +fields are resolved, so combine it with a large keg. Query predicates on the +same fields (`entity=plan`, `.hash=`) are evaluated by the Hub as well. ## Notes diff --git a/pkg/apicontract/contract.go b/pkg/apicontract/contract.go index 6c137c56..ad244647 100644 --- a/pkg/apicontract/contract.go +++ b/pkg/apicontract/contract.go @@ -15,7 +15,7 @@ import ( "time" ) -const Revision = "2026-09-11" +const Revision = "2026-09-28" const VersionHeader = "Tapper-API-Version" const ClientHeader = "Tapper-Client-Version" const Required = "API_VERSION_REQUIRED" diff --git a/pkg/apicontract/contract_test.go b/pkg/apicontract/contract_test.go index 73585844..d947b2a2 100644 --- a/pkg/apicontract/contract_test.go +++ b/pkg/apicontract/contract_test.go @@ -85,8 +85,8 @@ func TestClientDiscoveryMatrix(t *testing.T) { name, body, code string status int }{ - {"matching", `{"server_version":"0.24.0","api_versions":["2026-09-11"]}`, "", 200}, - {"newer compatible", `{"server_version":"99.0.0","api_versions":["2026-09-11","2099-01-01"]}`, "", 200}, + {"matching", `{"server_version":"0.24.0","api_versions":["2026-09-28"]}`, "", 200}, + {"newer compatible", `{"server_version":"99.0.0","api_versions":["2026-09-28","2099-01-01"]}`, "", 200}, {"newer incompatible", `{"server_version":"99.0.0","api_versions":["2099-01-01"]}`, Unsupported, 200}, {"legacy", "", DiscoveryUnavailable, 404}, {"redirect", "", DiscoveryUnavailable, 302}, {"malformed", "oops", DiscoveryInvalid, 200}, {"missing versions", `{"server_version":"v1"}`, DiscoveryInvalid, 200}, diff --git a/pkg/cli/cmd_mcp.go b/pkg/cli/cmd_mcp.go index 27d8dd8d..54e7bfcd 100644 --- a/pkg/cli/cmd_mcp.go +++ b/pkg/cli/cmd_mcp.go @@ -6,6 +6,7 @@ import ( "io" "log/slog" "strings" + "time" "github.com/jlrickert/cli-toolkit/mylog" "github.com/jlrickert/cli-toolkit/toolkit" @@ -89,9 +90,16 @@ func mcpServerOptions(logger *slog.Logger, reporter tapper.InvocationReporter) m // stdio puts the server on the same machine as its agent host, so // attachment paths in tool arguments name the same files for both sides. SharedFilesystem: true, + // Each orientation resolution costs several Hub requests; an agent's + // burst of tool calls reuses one for a few seconds (#123). + OrientationCacheTTL: mcpOrientationCacheTTL, } } +// mcpOrientationCacheTTL bounds how long `tap mcp` reuses a resolved +// orientation between tool calls. +const mcpOrientationCacheTTL = 10 * time.Second + // buildMCPLogger constructs the structured logger for the MCP server. // Unlike CLI commands, MCP always logs to stderr because stdout is reserved // for JSON-RPC. When a log file is also configured, entries fan out to both. diff --git a/pkg/keg/format_fields.go b/pkg/keg/format_fields.go index e394dd4e..c7d3d062 100644 --- a/pkg/keg/format_fields.go +++ b/pkg/keg/format_fields.go @@ -89,7 +89,7 @@ var LegacyFormatVerbs = map[byte]string{ // FormatVocabularyDescription is the one-line summary of the listing field // vocabulary. It is duplicated as a literal in the MCP tool schemas, which // require a constant struct tag; a test holds the two in agreement. -const FormatVocabularyDescription = "output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates read one file per node." +const FormatVocabularyDescription = "output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates are resolved by the Hub for the returned page in one batch." func isIndexTimeField(name string) bool { return slices.Contains(IndexTimeFieldNames, name) diff --git a/pkg/keg/keg_aggregate.go b/pkg/keg/keg_aggregate.go index 4f6ec8e7..28a755a2 100644 --- a/pkg/keg/keg_aggregate.go +++ b/pkg/keg/keg_aggregate.go @@ -41,6 +41,11 @@ type ListViewOptions struct { // it here rather than in the caller keeps paging and TotalMatches correct. TitleContains string `json:"title_contains,omitempty"` + // NodeIDs, when set, restricts the listing to these nodes. It is how a + // caller that already knows its result set (grep, links, tags) resolves + // the fields for all of it in one call instead of one read per node. + NodeIDs []NodeId `json:"node_ids,omitempty"` + // Fields are field selectors to resolve per row, in the vocabulary of // ParseFieldSelector ("type", ".omega", "tags"). Intrinsics and index // timestamps cost nothing; other selectors are read per returned row. @@ -82,6 +87,12 @@ type ReadNodesOptions struct { NodeIDs []NodeId `json:"node_ids,omitempty"` Query string `json:"query,omitempty"` Touch bool `json:"touch,omitempty"` + // MetaOnly reads each node's metadata and stats without its content or + // asset lists, in the same consistent read, so a listing that renders + // metadata fields never loads bodies. Content is still read for a node + // whose stats predate the state hash, to derive it. It cannot be combined + // with Touch: a metadata read is not someone opening the node. + MetaOnly bool `json:"meta_only,omitempty"` } type RelatedDirection string @@ -96,6 +107,22 @@ type RelatedNodesOptions struct { Direction RelatedDirection `json:"direction"` } +// RelatedPair is one relationship found by RelatedNodes: From is the input +// node's path and To is the related entry's ID exactly as it appears in +// Entries, which may be a qualified reference into another keg. +type RelatedPair struct { + From string `json:"from"` + To string `json:"to"` +} + +// RelatedNodesResult is the merged set of related entries plus the pairs that +// produced them, so callers can attribute each relationship to its input +// without asking once per input. +type RelatedNodesResult struct { + Entries []NodeIndexEntry `json:"entries"` + Pairs []RelatedPair `json:"pairs"` +} + type KegInfo struct { Settings *Settings `json:"settings"` Summary *KegSummary `json:"summary"` @@ -271,6 +298,19 @@ func (k *LocalKeg) listView(ctx context.Context, opts ListViewOptions) (*ListVie } entries = kept } + if len(opts.NodeIDs) > 0 { + wanted := make(map[string]struct{}, len(opts.NodeIDs)) + for _, id := range opts.NodeIDs { + wanted[id.Path()] = struct{}{} + } + kept := make([]NodeIndexEntry, 0, len(opts.NodeIDs)) + for _, entry := range entries { + if _, ok := wanted[nodeKey(entry)]; ok { + kept = append(kept, entry) + } + } + entries = kept + } total := len(entries) // Sorting by metadata or a non-timestamp stat needs a value for every @@ -486,6 +526,9 @@ func (k *LocalKeg) ReadNodes(ctx context.Context, opts ReadNodesOptions) ([]Node } func (k *LocalKeg) readNodes(ctx context.Context, opts ReadNodesOptions) ([]NodeView, error) { + if opts.MetaOnly && opts.Touch { + return nil, fmt.Errorf("meta_only and touch are mutually exclusive: %w", ErrInvalid) + } ids := slices.Clone(opts.NodeIDs) if q := strings.TrimSpace(opts.Query); q != "" { if len(ids) != 0 { @@ -526,9 +569,13 @@ func (k *LocalKeg) readNodes(ctx context.Context, opts ReadNodesOptions) ([]Node } } + read := k.ReadNode + if opts.MetaOnly { + read = k.readNodeMeta + } views := make([]NodeView, 0, len(ids)) for _, id := range ids { - view, err := k.ReadNode(ctx, id) + view, err := read(ctx, id) if err != nil { if errors.Is(err, ErrNotExist) { err = fmt.Errorf("node %s not found: %w", id.Path(), err) @@ -543,11 +590,11 @@ func (k *LocalKeg) readNodes(ctx context.Context, opts ReadNodesOptions) ([]Node return views, nil } -func (k *LocalKeg) RelatedNodes(ctx context.Context, opts RelatedNodesOptions) ([]NodeIndexEntry, error) { - return withKegReadValue(ctx, k, func(ctx context.Context) ([]NodeIndexEntry, error) { return k.relatedNodes(ctx, opts) }) +func (k *LocalKeg) RelatedNodes(ctx context.Context, opts RelatedNodesOptions) (*RelatedNodesResult, error) { + return withKegReadValue(ctx, k, func(ctx context.Context) (*RelatedNodesResult, error) { return k.relatedNodes(ctx, opts) }) } -func (k *LocalKeg) relatedNodes(ctx context.Context, opts RelatedNodesOptions) ([]NodeIndexEntry, error) { +func (k *LocalKeg) relatedNodes(ctx context.Context, opts RelatedNodesOptions) (*RelatedNodesResult, error) { if len(opts.NodeIDs) == 0 { return nil, fmt.Errorf("at least one node id is required: %w", ErrInvalid) } @@ -556,6 +603,8 @@ func (k *LocalKeg) relatedNodes(ctx context.Context, opts RelatedNodesOptions) ( return nil, err } seen := map[string]struct{}{} + pairs := []RelatedPair{} + pairSeen := map[RelatedPair]struct{}{} for _, id := range opts.NodeIDs { exists, err := k.NodeExists(ctx, id) if err != nil { @@ -575,6 +624,11 @@ func (k *LocalKeg) relatedNodes(ctx context.Context, opts RelatedNodesOptions) ( } for _, rel := range related { seen[rel.Path()] = struct{}{} + pair := RelatedPair{From: id.Path(), To: rel.Path()} + if _, dup := pairSeen[pair]; !dup { + pairSeen[pair] = struct{}{} + pairs = append(pairs, pair) + } } } entriesByID := map[string]NodeIndexEntry{} @@ -594,7 +648,7 @@ func (k *LocalKeg) relatedNodes(ctx context.Context, opts RelatedNodesOptions) ( out = append(out, NodeIndexEntry{ID: id}) } } - return out, nil + return &RelatedNodesResult{Entries: out, Pairs: pairs}, nil } func (k *LocalKeg) Info(ctx context.Context) (*KegInfo, error) { diff --git a/pkg/keg/keg_aggregate_test.go b/pkg/keg/keg_aggregate_test.go index 69be6d7f..0008f827 100644 --- a/pkg/keg/keg_aggregate_test.go +++ b/pkg/keg/keg_aggregate_test.go @@ -33,7 +33,8 @@ func TestLocalKegAggregateOperations(t *testing.T) { require.Equal(t, []string{"2", "1"}, []string{views[0].ID.Path(), views[1].ID.Path()}) related, err := k.RelatedNodes(ctx, keg.RelatedNodesOptions{NodeIDs: []keg.NodeId{two.ID}, Direction: keg.RelatedLinks}) require.NoError(t, err) - require.Equal(t, "1", related[0].ID) + require.Equal(t, "1", related.Entries[0].ID) + require.Equal(t, []keg.RelatedPair{{From: "2", To: "1"}}, related.Pairs) info, err := k.Info(ctx) require.NoError(t, err) require.Equal(t, 3, info.Summary.NodeCount) @@ -191,3 +192,43 @@ func TestRemoteAggregateMethodsUseOneRequest(t *testing.T) { }) } } + +func TestReadNodesMetaOnly(t *testing.T) { + fx := NewSandbox(t) + ctx := fx.Context() + repo := newTestMemoryRepo(fx.Runtime()) + k := keg.NewLocalKeg(repo, fx.Runtime()) + initNonStrictTestKeg(t, k, ctx) + created, err := k.Create(ctx, &keg.CreateOptions{Body: []byte("# One\n\nbody\n"), Meta: []byte("kind: note\n")}) + require.NoError(t, err) + id := created.ID + + full, err := k.ReadNode(ctx, id) + require.NoError(t, err) + views, err := k.ReadNodes(ctx, keg.ReadNodesOptions{NodeIDs: []keg.NodeId{id}, MetaOnly: true}) + require.NoError(t, err) + require.Len(t, views, 1) + require.Equal(t, full.Meta, views[0].Meta) + require.Equal(t, full.Hash(), views[0].Hash()) + require.Empty(t, views[0].Content) + + _, err = k.ReadNodes(ctx, keg.ReadNodesOptions{NodeIDs: []keg.NodeId{{ID: 999}}, MetaOnly: true}) + require.ErrorIs(t, err, keg.ErrNotExist) + + // A node that exists without metadata, and whose stats predate the state + // hash, is still a node: empty metadata and a hash derived from content. + bare, err := k.Create(ctx, &keg.CreateOptions{Body: []byte("# Bare\n\nbody\n")}) + require.NoError(t, err) + require.NoError(t, repo.WriteMeta(ctx, bare.ID, nil)) + require.NoError(t, repo.WriteStats(ctx, bare.ID, &keg.NodeStats{})) + bareFull, err := k.ReadNode(ctx, bare.ID) + require.NoError(t, err) + views, err = k.ReadNodes(ctx, keg.ReadNodesOptions{NodeIDs: []keg.NodeId{bare.ID}, MetaOnly: true}) + require.NoError(t, err) + require.Len(t, views, 1) + require.Empty(t, views[0].Meta) + require.NotEmpty(t, views[0].Hash()) + require.Equal(t, bareFull.Hash(), views[0].Hash()) + _, err = k.ReadNodes(ctx, keg.ReadNodesOptions{NodeIDs: []keg.NodeId{id}, MetaOnly: true, Touch: true}) + require.ErrorIs(t, err, keg.ErrInvalid) +} diff --git a/pkg/keg/keg_grep_projection_test.go b/pkg/keg/keg_grep_projection_test.go new file mode 100644 index 00000000..2e6e670d --- /dev/null +++ b/pkg/keg/keg_grep_projection_test.go @@ -0,0 +1,151 @@ +package keg_test + +import ( + "context" + "fmt" + "regexp" + "strings" + "testing" + + "github.com/jlrickert/tapper/pkg/keg" + "github.com/stretchr/testify/require" +) + +func seedGrepKeg(t *testing.T, repo keg.Repository) (keg.Keg, context.Context) { + t.Helper() + fx := NewSandbox(t) + ctx := fx.Context() + k := keg.NewLocalKeg(repo, fx.Runtime()) + initNonStrictTestKeg(t, k, ctx) + for _, body := range []string{ + "# Alpha\n\nneedle one\n", + "# Beta\n\nnothing here\n", + "# Gamma\n\nNEEDLE two\n", + "# Delta\n\nneedle three\n", + } { + _, err := k.Create(ctx, &keg.CreateOptions{Body: []byte(body), Meta: []byte("kind: note\n")}) + require.NoError(t, err) + } + require.NoError(t, k.Index(ctx, keg.IndexOptions{})) + return k, ctx +} + +func grepIDs(matches []keg.GrepMatch) []string { + ids := make([]string, len(matches)) + for i, m := range matches { + ids[i] = m.Entry.ID + } + return ids +} + +func TestGrepPagesAndProjectsFields(t *testing.T) { + fx := NewSandbox(t) + k, ctx := seedGrepKeg(t, newTestMemoryRepo(fx.Runtime())) + + all, err := k.Grep(ctx, keg.GrepOptions{Pattern: "needle", IgnoreCase: true}) + require.NoError(t, err) + require.Equal(t, []string{"1", "3", "4"}, grepIDs(all)) + + page, err := k.Grep(ctx, keg.GrepOptions{Pattern: "needle", IgnoreCase: true, Offset: 1, Limit: 1, Fields: []string{"kind"}}) + require.NoError(t, err) + require.Equal(t, []string{"3"}, grepIDs(page)) + require.Equal(t, map[string]string{"kind": "note"}, page[0].Fields) + + _, err = k.Grep(ctx, keg.GrepOptions{Pattern: "needle", Offset: -1}) + require.ErrorIs(t, err, keg.ErrInvalid) +} + +func TestListViewRestrictsToNodeIDs(t *testing.T) { + fx := NewSandbox(t) + k, ctx := seedGrepKeg(t, newTestMemoryRepo(fx.Runtime())) + view, err := k.ListView(ctx, keg.ListViewOptions{ + NodeIDs: []keg.NodeId{{ID: 4}, {ID: 2}}, + Fields: []string{"kind"}, + }) + require.NoError(t, err) + require.Len(t, view.Rows, 2) + require.Equal(t, "2", view.Rows[0].Entry.ID) + require.Equal(t, "4", view.Rows[1].Entry.ID) + require.Equal(t, "note", view.Rows[1].Fields["kind"]) + require.Equal(t, 2, view.TotalMatches) +} + +// grepRepo greps content itself, as a database backend would. Its matcher is +// Go's regexp, so results can be compared with the per-node path; patterns +// it cannot compile are reported as invalid, as the backend contract says. +type grepRepo struct { + keg.Repository + patterns []string +} + +func (r *grepRepo) GrepContent(ctx context.Context, opts keg.GrepContentOptions, fn func(keg.NodeId, []string) error) error { + r.patterns = append(r.patterns, opts.Pattern) + pattern := opts.Pattern + if opts.IgnoreCase { + pattern = "(?i)" + pattern + } + re, err := regexp.Compile(pattern) + if err != nil { + return fmt.Errorf("backend rejects %q: %w", opts.Pattern, keg.ErrInvalid) + } + ids, err := r.ListNodes(ctx) + if err != nil { + return err + } + for _, id := range ids { + raw, err := r.ReadContent(ctx, id) + if err != nil { + continue + } + var lines []string + for i, line := range strings.Split(strings.ReplaceAll(string(raw), "\r\n", "\n"), "\n") { + line = strings.TrimRight(line, "\r") + if re.MatchString(line) { + lines = append(lines, fmt.Sprintf("%d:%s", i+1, line)) + } + } + if opts.MaxLines > 0 && len(lines) > opts.MaxLines { + lines = lines[:opts.MaxLines] + } + if len(lines) > 0 { + if err := fn(id, lines); err != nil { + return err + } + } + } + return nil +} + +func TestGrepThroughBackendMatchesPlainGrep(t *testing.T) { + fx := NewSandbox(t) + plain, ctx := seedGrepKeg(t, newTestMemoryRepo(fx.Runtime())) + repo := &grepRepo{Repository: newTestMemoryRepo(fx.Runtime())} + backed, _ := seedGrepKeg(t, repo) + + for _, opts := range []keg.GrepOptions{ + {Pattern: "needle"}, + {Pattern: "needle", IgnoreCase: true}, + {Pattern: `\bneedle\s+t`}, + {Pattern: "(?i)NEEDLE two"}, + {Pattern: "nothing|needle"}, + {Pattern: "needle", MaxLines: 1}, + {Pattern: "needle", IgnoreCase: true, Offset: 1, Limit: 1, Fields: []string{"kind"}}, + } { + want, err := plain.Grep(ctx, opts) + require.NoError(t, err) + got, err := backed.Grep(ctx, opts) + require.NoError(t, err) + require.Equal(t, want, got, "pattern %q", opts.Pattern) + } +} + +// The backend owns its regex dialect: a pattern Go cannot compile still +// reaches it, and its refusal is reported as an invalid pattern. +func TestGrepLeavesPatternValidationToTheBackend(t *testing.T) { + fx := NewSandbox(t) + repo := &grepRepo{Repository: newTestMemoryRepo(fx.Runtime())} + k, ctx := seedGrepKeg(t, repo) + _, err := k.Grep(ctx, keg.GrepOptions{Pattern: "needle(?= one)"}) + require.ErrorIs(t, err, keg.ErrInvalid) + require.Equal(t, []string{"needle(?= one)"}, repo.patterns) +} diff --git a/pkg/keg/keg_iface.go b/pkg/keg/keg_iface.go index a54f1100..6db5dc62 100644 --- a/pkg/keg/keg_iface.go +++ b/pkg/keg/keg_iface.go @@ -189,7 +189,7 @@ type Keg interface { // RelatedNodes returns the deduplicated union of links or backlinks for the // supplied nodes, ordered by node id. It fails if no ids are supplied, an id // is missing, or the direction is invalid. - RelatedNodes(ctx context.Context, opts RelatedNodesOptions) ([]NodeIndexEntry, error) + RelatedNodes(ctx context.Context, opts RelatedNodesOptions) (*RelatedNodesResult, error) // Doctor inspects configuration, content, links, metadata, stats, and schema // validation and returns deterministic diagnostic issues without mutating the @@ -348,18 +348,28 @@ type QueryOptions struct { // GrepOptions configures Keg.Grep. type GrepOptions struct { // Pattern is a Go regular expression matched against content lines. - Pattern string + Pattern string `json:"pattern"` // IgnoreCase makes the match case-insensitive. - IgnoreCase bool + IgnoreCase bool `json:"ignore_case,omitempty"` // MaxLines caps matched lines per node. 0 means no cap. - MaxLines int + MaxLines int `json:"max_lines,omitempty"` + // Fields are field selectors to resolve for each returned match, in the + // vocabulary of ParseFieldSelector. They are resolved after paging, in + // one batch, so a formatted grep never reads nodes one at a time. + Fields []string `json:"fields,omitempty"` + // Offset skips the first N matching nodes; Limit caps the matches + // returned (0 means no limit). Both apply before field projection. + Offset int `json:"offset,omitempty"` + Limit int `json:"limit,omitempty"` } // GrepMatch reports one node's content matches for Keg.Grep. type GrepMatch struct { - Entry NodeIndexEntry + Entry NodeIndexEntry `json:"entry"` // Lines are rendered "lineno:text" match lines. - Lines []string + Lines []string `json:"lines"` + // Fields holds the requested selectors' values keyed by selector text. + Fields map[string]string `json:"fields,omitempty"` } // AssetSummary aggregates one asset kind for KegSummary. diff --git a/pkg/keg/keg_local_view.go b/pkg/keg/keg_local_view.go index 9ba0c323..39791b98 100644 --- a/pkg/keg/keg_local_view.go +++ b/pkg/keg/keg_local_view.go @@ -13,6 +13,49 @@ func (k *LocalKeg) ReadNode(ctx context.Context, id NodeId) (*NodeView, error) { return withKegReadValue(ctx, k, func(ctx context.Context) (*NodeView, error) { return k.readNode(ctx, id) }) } +// readNodeMeta is readNode without content or asset lists: raw meta (empty +// when absent), stats, and the state hash. Content is read only for a node +// whose stats carry no hash, because that hash is derived from content. The +// caller holds the read operation. +func (k *LocalKeg) readNodeMeta(ctx context.Context, id NodeId) (*NodeView, error) { + if err := k.checkKegExists(ctx); err != nil { + return nil, fmt.Errorf("failed to read node meta: %w", err) + } + exists, err := k.Repo.HasNode(ctx, id) + if err != nil { + return nil, err + } + if !exists { + return nil, ErrNotExist + } + meta, err := k.Repo.ReadMeta(ctx, id) + if err != nil && !errors.Is(err, ErrNotExist) { + return nil, err + } + stats, err := k.Repo.ReadStats(ctx, id) + if err != nil && !errors.Is(err, ErrNotExist) { + return nil, err + } + if stats == nil { + stats = &NodeStats{} + } + view := &NodeView{ID: id, Meta: meta, Stats: stats} + if stats.Hash() == "" { + content, err := k.Repo.ReadContent(ctx, id) + if err != nil { + return nil, err + } + parsedContent, parseErr := ParseContent(k.Runtime, content, MarkdownContentFilename) + if parseErr == nil { + parsedMeta, metaErr := ParseMeta(ctx, meta) + if metaErr == nil { + view.hash = nodeStateHash(k.Runtime, parsedContent.Hash, parsedMeta) + } + } + } + return view, nil +} + func (k *LocalKeg) readNode(ctx context.Context, id NodeId) (*NodeView, error) { if err := k.checkKegExists(ctx); err != nil { return nil, fmt.Errorf("failed to read node: %w", err) diff --git a/pkg/keg/keg_query.go b/pkg/keg/keg_query.go index de632f3b..baab68e6 100644 --- a/pkg/keg/keg_query.go +++ b/pkg/keg/keg_query.go @@ -53,41 +53,98 @@ func (k *LocalKeg) Grep(ctx context.Context, opts GrepOptions) ([]GrepMatch, err } func (k *LocalKeg) grep(ctx context.Context, opts GrepOptions) ([]GrepMatch, error) { - pattern := opts.Pattern - if opts.IgnoreCase { - pattern = "(?i)" + pattern + if opts.Offset < 0 || opts.Limit < 0 { + return nil, fmt.Errorf("grep offset and limit must be >= 0: %w", ErrInvalid) } - re, err := regexp.Compile(pattern) + selectors, err := ParseFieldSelectors(opts.Fields) if err != nil { - return nil, fmt.Errorf("invalid grep pattern %q: %w", opts.Pattern, err) + return nil, err } - dex, err := k.Dex(ctx) if err != nil { return nil, fmt.Errorf("unable to read dex: %w", err) } + // A backend that greps itself matches in its own dialect and returns + // only matching lines, so content never has to be read out. Otherwise + // each node is read and matched with Go's regexp. + var matchedBy func(NodeId) ([]string, error) + if backend, ok := k.Repo.(RepositoryGrep); ok { + found := map[string][]string{} + err := backend.GrepContent(ctx, GrepContentOptions{Pattern: opts.Pattern, IgnoreCase: opts.IgnoreCase, MaxLines: opts.MaxLines}, + func(id NodeId, lines []string) error { + if len(lines) > 0 { + found[id.Path()] = lines + } + return nil + }) + if err != nil { + if errors.Is(err, ErrInvalid) { + return nil, fmt.Errorf("invalid grep pattern %q: %w", opts.Pattern, err) + } + return nil, fmt.Errorf("unable to search node content: %w", err) + } + matchedBy = func(id NodeId) ([]string, error) { return found[id.Path()], nil } + } else { + pattern := opts.Pattern + if opts.IgnoreCase { + pattern = "(?i)" + pattern + } + re, err := regexp.Compile(pattern) + if err != nil { + return nil, fmt.Errorf("invalid grep pattern %q: %v: %w", opts.Pattern, err, ErrInvalid) + } + matchedBy = func(id NodeId) ([]string, error) { + raw, err := k.Repo.ReadContent(ctx, id) + if err != nil { + if errors.Is(err, ErrNotExist) { + return nil, nil + } + return nil, fmt.Errorf("unable to read node content: %w", err) + } + lines := grepContentLineMatches(re, raw) + if opts.MaxLines > 0 && len(lines) > opts.MaxLines { + lines = lines[:opts.MaxLines] + } + return lines, nil + } + } + matches := make([]GrepMatch, 0) for _, entry := range dex.Nodes(ctx) { id, parseErr := ParseNode(entry.ID) if parseErr != nil || id == nil { continue } - raw, readErr := k.Repo.ReadContent(ctx, *id) - if readErr != nil { - if errors.Is(readErr, ErrNotExist) { - continue - } - return nil, fmt.Errorf("unable to read node content: %w", readErr) - } - lines := grepContentLineMatches(re, raw) - if opts.MaxLines > 0 && len(lines) > opts.MaxLines { - lines = lines[:opts.MaxLines] + lines, err := matchedBy(*id) + if err != nil { + return nil, err } if len(lines) > 0 { matches = append(matches, GrepMatch{Entry: entry, Lines: lines}) } } + + if opts.Offset > 0 { + if opts.Offset >= len(matches) { + matches = matches[:0] + } else { + matches = matches[opts.Offset:] + } + } + if opts.Limit > 0 && len(matches) > opts.Limit { + matches = matches[:opts.Limit] + } + if len(selectors) > 0 && len(matches) > 0 { + entries := make([]NodeIndexEntry, len(matches)) + for i, m := range matches { + entries[i] = m.Entry + } + values := k.loadFieldValues(ctx, entries, selectors) + for i := range matches { + matches[i].Fields = resolveRowFields(matches[i].Entry, selectors, values) + } + } return matches, nil } diff --git a/pkg/keg/keg_remote.go b/pkg/keg/keg_remote.go index 16086334..403d774d 100644 --- a/pkg/keg/keg_remote.go +++ b/pkg/keg/keg_remote.go @@ -137,6 +137,49 @@ func (k *RemoteKeg) SetTarget(target *Target) { // do executes an HTTP request with authentication and context propagation. func (k *RemoteKeg) do(ctx context.Context, method, path string, body io.Reader, contentType string, header http.Header) (*http.Response, error) { + return k.doURL(ctx, method, k.baseURL+path, path, body, contentType, header) +} + +// hubAPIRoot is the Hub's /api/v1 root this keg's routes live under. +func (k *RemoteKeg) hubAPIRoot() (string, error) { + const marker = "/api/v1/" + i := strings.Index(k.baseURL, marker) + if i < 0 { + return "", NewBackendError("remote", "hub root", 0, fmt.Errorf("keg URL %q is not under %s: %w", k.baseURL, marker, ErrInvalid), false) + } + return k.baseURL[:i+len(marker)-1], nil +} + +// HubPost sends a JSON request to a path under this keg's Hub API root (for +// example "/tap/links") with the same credentials, contract negotiation, and +// orientation as the keg's own routes, and decodes the JSON response into out. +func (k *RemoteKeg) HubPost(ctx context.Context, path, op string, in, out any) error { + root, err := k.hubAPIRoot() + if err != nil { + return err + } + payload, err := json.Marshal(in) + if err != nil { + return NewBackendError("remote", op, 0, err, false) + } + resp, err := k.doURL(ctx, http.MethodPost, root+path, path, bytes.NewReader(payload), "application/json", nil) + if err != nil { + return err + } + body, err := k.readBody(resp, op, http.StatusOK) + if err != nil { + return err + } + if out == nil { + return nil + } + if err := json.Unmarshal(body, out); err != nil { + return NewBackendError("remote", op, 0, err, false) + } + return nil +} + +func (k *RemoteKeg) doURL(ctx context.Context, method, url, path string, body io.Reader, contentType string, header http.Header) (*http.Response, error) { if apicontract.FromContext(ctx) == nil { ctx = apicontract.WithSession(ctx, k.contract) } @@ -151,7 +194,7 @@ func (k *RemoteKeg) do(ctx context.Context, method, path string, body io.Reader, if err := ValidateOrientationTarget(ctx, k.baseURL); err != nil { return nil, err } - req, err := http.NewRequestWithContext(ctx, method, k.baseURL+path, body) + req, err := http.NewRequestWithContext(ctx, method, url, body) if err != nil { return nil, NewBackendError("remote", method+" "+path, 0, err, false) } @@ -515,20 +558,16 @@ func parseRewritten(paths []string) ([]NodeId, error) { return out, nil } -// Move implements Keg via POST /nodes/{src}/move. +// Move implements Keg as a batch-of-one call to POST /nodes/move. func (k *RemoteKeg) Move(ctx context.Context, opts NodeMoveOptions) ([]NodeId, error) { - var result struct { - Rewritten []string `json:"rewritten"` - } - req := struct { - Dst int `json:"dst"` - ExpectedHash string `json:"expected_hash"` - }{Dst: opts.Destination.ID, ExpectedHash: opts.ExpectedHash} - path := fmt.Sprintf("/nodes/%d/move", opts.Source.ID) - if err := k.postJSON(ctx, path, "Move", req, &result, http.StatusOK); err != nil { + results, err := k.MoveBatch(ctx, []MoveItem{{Source: opts.Source.ID, Destination: opts.Destination.ID, ExpectedHash: opts.ExpectedHash}}) + if err != nil { return nil, err } - return parseRewritten(result.Rewritten) + if len(results) != 1 { + return nil, NewBackendError("remote", "Move", 0, fmt.Errorf("invalid response: expected one moved node, got %d", len(results)), false) + } + return results[0].Rewritten, nil } // Remove implements Keg as a batch-of-one call to POST /nodes/remove. @@ -688,27 +727,19 @@ func (k *RemoteKeg) Query(ctx context.Context, opts QueryOptions) ([]NodeIndexEn return result.Entries, nil } -// Grep implements Keg via POST /grep. +// Grep implements Keg via POST /grep. Paging and field projection happen on +// the server, so a formatted grep is one request regardless of result size. func (k *RemoteKeg) Grep(ctx context.Context, opts GrepOptions) ([]GrepMatch, error) { var result struct { - Matches []struct { - Entry NodeIndexEntry `json:"entry"` - Lines []string `json:"lines"` - } `json:"matches"` + Matches []GrepMatch `json:"matches"` } - req := struct { - Pattern string `json:"pattern"` - IgnoreCase bool `json:"ignore_case"` - MaxLines int `json:"max_lines"` - }{Pattern: opts.Pattern, IgnoreCase: opts.IgnoreCase, MaxLines: opts.MaxLines} - if err := k.postJSON(ctx, "/grep", "Grep", req, &result, http.StatusOK); err != nil { + if err := k.postJSON(ctx, "/grep", "Grep", opts, &result, http.StatusOK); err != nil { return nil, err } - out := make([]GrepMatch, len(result.Matches)) - for i, m := range result.Matches { - out[i] = GrepMatch{Entry: m.Entry, Lines: m.Lines} + if result.Matches == nil { + result.Matches = []GrepMatch{} } - return out, nil + return result.Matches, nil } // Index implements Keg via POST /indexes/rebuild. @@ -748,83 +779,85 @@ func (k *RemoteKeg) Summary(ctx context.Context) (*KegSummary, error) { // --- Assets and images --- -// listAssets fetches an asset name list (assets or images) for a node. -func (k *RemoteKeg) listAssets(ctx context.Context, id NodeId, kind, op string) ([]string, error) { - var names []string - if err := k.getJSON(ctx, fmt.Sprintf("/nodes/%d/%s", id.ID, kind), op, &names); err != nil { - return nil, err +// attachmentKind maps the legacy asset families onto attachment kinds. +func attachmentKind(kind string) AttachmentKind { + if kind == "images" { + return AttachmentImage } - return names, nil + return AttachmentFile } -// readAsset fetches one asset payload (asset or image) for a node. -func (k *RemoteKeg) readAsset(ctx context.Context, id NodeId, kind, name, op string) ([]byte, error) { - path := fmt.Sprintf("/nodes/%d/%s/%s", id.ID, kind, url.PathEscape(name)) - resp, err := k.do(ctx, http.MethodGet, path, nil, "", nil) - if err != nil { - return nil, err +// listAssets lists one attachment kind for a node through the attachment +// listing, following its cursor to the end. +func (k *RemoteKeg) listAssets(ctx context.Context, id NodeId, kind, op string) ([]string, error) { + want := attachmentKind(kind) + names := []string{} + cursor := "" + for { + page, err := k.ListAttachments(ctx, AttachmentListRequest{IDs: []int{id.ID}, Cursor: cursor, Limit: 100}) + if err != nil { + return nil, err + } + for _, item := range page.Results { + if item.Kind == want { + names = append(names, item.Name) + } + } + if page.Cursor == "" || page.Cursor == cursor { + return names, nil + } + cursor = page.Cursor } - return k.readBody(resp, op, http.StatusOK) } -// writeAsset stores one asset payload (asset or image) for a node. -func (k *RemoteKeg) writeAsset(ctx context.Context, id NodeId, kind, name string, data []byte, op string) error { - path := fmt.Sprintf("/nodes/%d/%s/%s", id.ID, kind, url.PathEscape(name)) - resp, err := k.do(ctx, http.MethodPut, path, bytes.NewReader(data), "application/octet-stream", nil) - if err != nil { - return err - } - _, err = k.readBody(resp, op, http.StatusOK, http.StatusNoContent) - return err +func (k *RemoteKeg) readAsset(ctx context.Context, id NodeId, kind, name, _ string) ([]byte, error) { + return k.ReadAttachment(ctx, id, attachmentKind(kind), name) } -// deleteAsset removes one asset (asset or image) from a node. -func (k *RemoteKeg) deleteAsset(ctx context.Context, id NodeId, kind, name, op string) error { - path := fmt.Sprintf("/nodes/%d/%s/%s", id.ID, kind, url.PathEscape(name)) - resp, err := k.do(ctx, http.MethodDelete, path, nil, "", nil) - if err != nil { - return err - } - _, err = k.readBody(resp, op, http.StatusOK, http.StatusNoContent) - return err +func (k *RemoteKeg) writeAsset(ctx context.Context, id NodeId, kind, name string, data []byte, _ string) error { + return k.WriteAttachment(ctx, id, attachmentKind(kind), name, data) } -// ListFiles implements Keg via GET /nodes/{id}/assets. +func (k *RemoteKeg) deleteAsset(ctx context.Context, id NodeId, kind, name, _ string) error { + return k.DeleteAttachment(ctx, id, attachmentKind(kind), name) +} + +// ListFiles implements Keg through the attachment listing. func (k *RemoteKeg) ListFiles(ctx context.Context, id NodeId) ([]string, error) { return k.listAssets(ctx, id, "assets", "ListFiles") } -// ReadFile implements Keg via GET /nodes/{id}/assets/{name}. +// ReadFile implements Keg through the file attachment route. func (k *RemoteKeg) ReadFile(ctx context.Context, id NodeId, name string) ([]byte, error) { return k.readAsset(ctx, id, "assets", name, "ReadFile") } -// WriteFile implements Keg via PUT /nodes/{id}/assets/{name}. +// WriteFile implements Keg through the file attachment route. func (k *RemoteKeg) WriteFile(ctx context.Context, id NodeId, name string, data []byte) error { return k.writeAsset(ctx, id, "assets", name, data, "WriteFile") } -// DeleteFile implements Keg via DELETE /nodes/{id}/assets/{name}. +// DeleteFile implements Keg through the file attachment route. func (k *RemoteKeg) DeleteFile(ctx context.Context, id NodeId, name string) error { return k.deleteAsset(ctx, id, "assets", name, "DeleteFile") } -// ListImages implements Keg via GET /nodes/{id}/images. +// ListImages implements Keg through the attachment listing. func (k *RemoteKeg) ListImages(ctx context.Context, id NodeId) ([]string, error) { return k.listAssets(ctx, id, "images", "ListImages") } -// ReadImage implements Keg via GET /nodes/{id}/images/{name}. +// ReadImage implements Keg through the image attachment route. func (k *RemoteKeg) ReadImage(ctx context.Context, id NodeId, name string) ([]byte, error) { return k.readAsset(ctx, id, "images", name, "ReadImage") } -// WriteImage implements Keg via PUT /nodes/{id}/images/{name}. +// WriteImage implements Keg through the image attachment route. func (k *RemoteKeg) WriteImage(ctx context.Context, id NodeId, name string, data []byte) error { return k.writeAsset(ctx, id, "images", name, data, "WriteImage") } -// DeleteImage implements Keg via DELETE /nodes/{id}/images/{name}. +// DeleteImage implements Keg through the image attachment route. func (k *RemoteKeg) DeleteImage(ctx context.Context, id NodeId, name string) error { return k.deleteAsset(ctx, id, "images", name, "DeleteImage") } @@ -942,10 +975,29 @@ func (k *RemoteKeg) ReadContentAt(ctx context.Context, id NodeId, rev RevisionID return k.readBody(resp, "ReadContentAt", http.StatusOK) } -// RestoreSnapshot implements Keg via POST /nodes/{id}/snapshots/{rev}/restore. +// RestoreSnapshot implements Keg as a batch-of-one call to POST +// /nodes/snapshots/restore, guarded by the node's current hash. func (k *RemoteKeg) RestoreSnapshot(ctx context.Context, id NodeId, rev RevisionID) error { - path := fmt.Sprintf("/nodes/%d/snapshots/%d/restore", id.ID, int64(rev)) - return k.postJSON(ctx, path, "RestoreSnapshot", struct{}{}, nil, http.StatusOK, http.StatusNoContent) + hash, err := k.currentHash(ctx, id) + if err != nil { + return err + } + _, err = k.RestoreBatch(ctx, []RestoreItem{{ID: id.ID, Revision: rev, ExpectedHash: hash}}) + return err +} + +// currentHash reads a node's precondition hash from its metadata route's +// ETag, which neither loads content nor counts as a read. +func (k *RemoteKeg) currentHash(ctx context.Context, id NodeId) (string, error) { + resp, err := k.do(ctx, http.MethodGet, fmt.Sprintf("/nodes/%d/meta", id.ID), nil, "", nil) + if err != nil { + return "", err + } + etag := resp.Header.Get("ETag") + if _, err := k.readBody(resp, "RestoreSnapshot", http.StatusOK); err != nil { + return "", err + } + return strings.Trim(etag, `"`), nil } // --- Bulk transfer --- diff --git a/pkg/keg/keg_remote_aggregate.go b/pkg/keg/keg_remote_aggregate.go index a39d24e1..d5bcaa07 100644 --- a/pkg/keg/keg_remote_aggregate.go +++ b/pkg/keg/keg_remote_aggregate.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "encoding/json" - "errors" "fmt" "net/http" ) @@ -17,19 +16,10 @@ func (k *RemoteKeg) ListEntries(ctx context.Context, opts ListEntriesOptions) (* return &out, nil } -// ErrListViewUnsupported reports that the hub predates the server-resolved -// listing endpoint. Callers degrade to assembling the listing client-side. -var ErrListViewUnsupported = errors.New("hub list view API is unavailable") - -// ListView resolves a whole listing page in one request. A hub that does not -// implement the route answers 404, which is reported as -// ErrListViewUnsupported so the caller can fall back rather than fail. +// ListView resolves a whole listing page in one request. func (k *RemoteKeg) ListView(ctx context.Context, opts ListViewOptions) (*ListViewResult, error) { var out ListViewResult if err := k.postJSON(ctx, "/list/view", "ListView", opts, &out, http.StatusOK); err != nil { - if _, status := RemoteErrorCode(err); status == http.StatusNotFound { - return nil, fmt.Errorf("%w: %w", ErrListViewUnsupported, err) - } return nil, err } return &out, nil @@ -65,10 +55,11 @@ func (k *RemoteKeg) ReadNodes(ctx context.Context, opts ReadNodesOptions) ([]Nod ids[i] = id.ID } req := struct { - NodeIDs []int `json:"node_ids,omitempty"` - Query string `json:"query,omitempty"` - Touch bool `json:"touch,omitempty"` - }{ids, opts.Query, opts.Touch} + NodeIDs []int `json:"node_ids,omitempty"` + Query string `json:"query,omitempty"` + Touch bool `json:"touch,omitempty"` + MetaOnly bool `json:"meta_only,omitempty"` + }{ids, opts.Query, opts.Touch, opts.MetaOnly} var wire []remoteNodeView if err := k.postJSON(ctx, "/nodes/read", "ReadNodes", req, &wire, http.StatusOK); err != nil { return nil, err @@ -84,7 +75,7 @@ func (k *RemoteKeg) ReadNodes(ctx context.Context, opts ReadNodesOptions) ([]Nod return out, nil } -func (k *RemoteKeg) RelatedNodes(ctx context.Context, opts RelatedNodesOptions) ([]NodeIndexEntry, error) { +func (k *RemoteKeg) RelatedNodes(ctx context.Context, opts RelatedNodesOptions) (*RelatedNodesResult, error) { ids := make([]int, len(opts.NodeIDs)) for i, id := range opts.NodeIDs { ids[i] = id.ID @@ -93,13 +84,17 @@ func (k *RemoteKeg) RelatedNodes(ctx context.Context, opts RelatedNodesOptions) NodeIDs []int `json:"node_ids"` Direction RelatedDirection `json:"direction"` }{ids, opts.Direction} - var out struct { - Entries []NodeIndexEntry `json:"entries"` - } + var out RelatedNodesResult if err := k.postJSON(ctx, "/related", "RelatedNodes", req, &out, http.StatusOK); err != nil { return nil, err } - return out.Entries, nil + if out.Entries == nil { + out.Entries = []NodeIndexEntry{} + } + if out.Pairs == nil { + out.Pairs = []RelatedPair{} + } + return &out, nil } func (k *RemoteKeg) Info(ctx context.Context) (*KegInfo, error) { diff --git a/pkg/keg/keg_remote_listview_test.go b/pkg/keg/keg_remote_listview_test.go index c321b0df..f121a6be 100644 --- a/pkg/keg/keg_remote_listview_test.go +++ b/pkg/keg/keg_remote_listview_test.go @@ -92,22 +92,6 @@ func TestRemoteListViewIsOneRoundTrip(t *testing.T) { } } -// TestRemoteListViewUnsupportedIsDetectable proves a hub that predates the -// route is reported distinctly, so callers degrade to assembling the listing -// themselves instead of surfacing a bare 404. -func TestRemoteListViewUnsupportedIsDetectable(t *testing.T) { - t.Parallel() - - rec := newPathRecorder(t, func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusNotFound) - }) - - k := newRecorderKeg(t, rec) - _, err := k.ListView(context.Background(), kegpkg.ListViewOptions{Fields: []string{"type"}}) - require.Error(t, err) - require.ErrorIs(t, err, kegpkg.ErrListViewUnsupported) -} - func itoa(i int) string { if i == 0 { return "0" diff --git a/pkg/keg/keg_remote_test.go b/pkg/keg/keg_remote_test.go index b4141e11..bee395de 100644 --- a/pkg/keg/keg_remote_test.go +++ b/pkg/keg/keg_remote_test.go @@ -299,25 +299,20 @@ func newMockOpsHub(t *testing.T, f *sandbox.Sandbox, token string) *mockOpsHub { } h.writeJSON(w, http.StatusOK, response) }) - mux.HandleFunc("POST /nodes/{id}/move", func(w http.ResponseWriter, r *http.Request) { - id, ok := h.parseID(w, r) - if !ok { - return - } + mux.HandleFunc("POST /nodes/move", func(w http.ResponseWriter, r *http.Request) { var req struct { - Dst int `json:"dst"` - ExpectedHash string `json:"expected_hash"` + Nodes []kegpkg.MoveItem `json:"nodes"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { h.writeError(w, http.StatusBadRequest, "invalid JSON body", "BAD_REQUEST") return } - rewritten, err := backing.Move(r.Context(), kegpkg.NodeMoveOptions{Source: id, Destination: kegpkg.NodeId{ID: req.Dst}, ExpectedHash: req.ExpectedHash}) + results, err := backing.MoveBatch(r.Context(), req.Nodes) if err != nil { h.kegError(w, err) return } - h.writeJSON(w, http.StatusOK, map[string][]string{"rewritten": rewrittenWire(rewritten)}) + h.writeJSON(w, http.StatusOK, results) }) mux.HandleFunc("POST /nodes/{id}/touch", func(w http.ResponseWriter, r *http.Request) { id, ok := h.parseID(w, r) diff --git a/pkg/keg/orientation_test.go b/pkg/keg/orientation_test.go index d42175ba..020b330a 100644 --- a/pkg/keg/orientation_test.go +++ b/pkg/keg/orientation_test.go @@ -13,7 +13,7 @@ type orientationRoundTripFunc func(*http.Request) (*http.Response, error) func (f orientationRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { if req.URL.Path == "/api/version" { - return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"server_version":"test","api_versions":["2026-09-11"]}`))}, nil + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"server_version":"test","api_versions":["2026-09-28"]}`))}, nil } return f(req) } diff --git a/pkg/keg/repository.go b/pkg/keg/repository.go index 0ea36e43..7257bec5 100644 --- a/pkg/keg/repository.go +++ b/pkg/keg/repository.go @@ -169,6 +169,28 @@ func repositoryBatchRead(repo Repository) (RepositoryBatchRead, bool) { return capability, ok } +// GrepContentOptions is a content search a backend runs itself. +type GrepContentOptions struct { + // Pattern is a regular expression in the backend's own dialect. + Pattern string + // IgnoreCase matches case-insensitively. + IgnoreCase bool + // MaxLines caps matched lines per node; 0 means no cap. + MaxLines int +} + +// RepositoryGrep is an optional capability that lets a backend match content +// itself, in its own regular-expression dialect, so content never has to be +// read out to be searched. It reports a pattern it cannot evaluate as +// ErrInvalid. +type RepositoryGrep interface { + // GrepContent calls fn once per node with a matching line, in no + // particular node order, and stops at the first error fn returns. Lines + // are "lineno:text" in line order: 1-based numbers over content split at + // "\n" after "\r\n" becomes "\n", with a trailing "\r" trimmed. + GrepContent(ctx context.Context, opts GrepContentOptions, fn func(id NodeId, lines []string) error) error +} + // RepositoryFiles provides optional per-node file attachment access. type RepositoryFiles interface { // ListFiles lists file attachment names for a node. diff --git a/pkg/keg/surface_batch.go b/pkg/keg/surface_batch.go index 78bc840a..0a927f96 100644 --- a/pkg/keg/surface_batch.go +++ b/pkg/keg/surface_batch.go @@ -33,13 +33,16 @@ type MutationResult struct { Rewritten []NodeId `json:"rewritten,omitempty"` } -func (k *LocalKeg) preflightMutation(ctx context.Context, ids []int, hashes []string) error { +// preflightMutation checks a batch's nodes exist at the expected versions. +// The root (node 0) is accepted only when allowRoot is set: it can be restored +// from its own history but never moved or removed. +func (k *LocalKeg) preflightMutation(ctx context.Context, ids []int, hashes []string, allowRoot bool) error { if err := validateMutationBatchSize(len(ids)); err != nil { return err } seen := map[int]bool{} for i, id := range ids { - if id <= 0 || seen[id] { + if id < 0 || (id == 0 && !allowRoot) || seen[id] { return fmt.Errorf("duplicate or invalid node %d: %w", id, ErrInvalid) } seen[id] = true @@ -66,7 +69,7 @@ func (k *LocalKeg) MoveBatch(ctx context.Context, items []MoveItem) ([]MutationR } destinations[item.Destination] = true } - if err := k.preflightMutation(ctx, ids, hashes); err != nil { + if err := k.preflightMutation(ctx, ids, hashes, false); err != nil { return nil, err } // Check all destinations before any move so an earlier move cannot vacate a @@ -114,7 +117,7 @@ func (k *LocalKeg) RemoveBatch(ctx context.Context, items []RemoveItem) ([]Mutat for i, item := range items { ids[i], hashes[i] = item.ID, item.ExpectedHash } - if err := k.preflightMutation(ctx, ids, hashes); err != nil { + if err := k.preflightMutation(ctx, ids, hashes, false); err != nil { return nil, err } out := make([]MutationResult, 0, len(items)) @@ -140,7 +143,7 @@ func (k *LocalKeg) RestoreBatch(ctx context.Context, items []RestoreItem) ([]Mut for i, item := range items { ids[i], hashes[i] = item.ID, item.ExpectedHash } - if err := k.preflightMutation(ctx, ids, hashes); err != nil { + if err := k.preflightMutation(ctx, ids, hashes, true); err != nil { return nil, err } out := make([]MutationResult, 0, len(items)) @@ -180,10 +183,25 @@ func (k *RemoteKeg) MoveBatch(ctx context.Context, items []MoveItem) ([]Mutation err := k.postJSON(ctx, "/nodes/move", "MoveBatch", map[string]any{"nodes": items}, &out, http.StatusOK) return out, err } +// RemoveBatch implements Keg through POST /nodes/remove, the Hub's single +// removal contract: all items are removed atomically or none are. func (k *RemoteKeg) RemoveBatch(ctx context.Context, items []RemoveItem) ([]MutationResult, error) { - var out []MutationResult - err := k.postJSON(ctx, "/nodes/remove", "RemoveBatch", map[string]any{"nodes": items}, &out, http.StatusOK) - return out, err + nodes := make([]NodeRemoveOptions, len(items)) + for i, item := range items { + nodes[i] = NodeRemoveOptions{ID: NodeId{ID: item.ID}, ExpectedHash: item.ExpectedHash} + } + result, err := k.RemoveNodes(ctx, RemoveNodesOptions{Nodes: nodes}) + if err != nil { + return nil, err + } + if result.Failure != nil { + return nil, result.Failure.Err() + } + out := make([]MutationResult, len(result.Removed)) + for i, removed := range result.Removed { + out[i] = MutationResult{ID: removed.ID.ID, Rewritten: removed.Rewritten} + } + return out, nil } func (k *RemoteKeg) RestoreBatch(ctx context.Context, items []RestoreItem) ([]MutationResult, error) { var out []MutationResult diff --git a/pkg/keg/surface_batch_root_test.go b/pkg/keg/surface_batch_root_test.go new file mode 100644 index 00000000..d6beba7a --- /dev/null +++ b/pkg/keg/surface_batch_root_test.go @@ -0,0 +1,40 @@ +package keg_test + +import ( + "testing" + + "github.com/jlrickert/tapper/pkg/keg" + "github.com/stretchr/testify/require" +) + +// The root can be restored from its own history, but never moved or removed. +func TestBatchMutationsAndTheRoot(t *testing.T) { + fx := NewSandbox(t) + ctx := fx.Context() + k := keg.NewLocalKeg(newTestMemoryRepo(fx.Runtime()), fx.Runtime()) + initNonStrictTestKeg(t, k, ctx) + + root, err := k.ReadNode(ctx, keg.NodeId{ID: 0}) + require.NoError(t, err) + _, err = k.UpdateNodes(ctx, []keg.NodeUpdateOptions{{ID: keg.NodeId{ID: 0}, Content: []byte("# Root v1\n"), HasContent: true, ExpectedHash: root.Hash()}}) + require.NoError(t, err) + snaps, err := k.AppendSnapshots(ctx, []keg.NodeSnapshotRequest{{ID: keg.NodeId{ID: 0}, Message: "v1"}}) + require.NoError(t, err) + root, err = k.ReadNode(ctx, keg.NodeId{ID: 0}) + require.NoError(t, err) + _, err = k.UpdateNodes(ctx, []keg.NodeUpdateOptions{{ID: keg.NodeId{ID: 0}, Content: []byte("# Root v2\n"), HasContent: true, ExpectedHash: root.Hash()}}) + require.NoError(t, err) + root, err = k.ReadNode(ctx, keg.NodeId{ID: 0}) + require.NoError(t, err) + + _, err = k.MoveBatch(ctx, []keg.MoveItem{{Source: 0, Destination: 9, ExpectedHash: root.Hash()}}) + require.ErrorIs(t, err, keg.ErrInvalid) + _, err = k.RemoveBatch(ctx, []keg.RemoveItem{{ID: 0, ExpectedHash: root.Hash()}}) + require.ErrorIs(t, err, keg.ErrInvalid) + + _, err = k.RestoreBatch(ctx, []keg.RestoreItem{{ID: 0, Revision: snaps[0].ID, ExpectedHash: root.Hash()}}) + require.NoError(t, err) + content, err := k.GetContent(ctx, keg.NodeId{ID: 0}) + require.NoError(t, err) + require.Contains(t, string(content), "Root v1") +} diff --git a/pkg/mcp/orientation_cache_test.go b/pkg/mcp/orientation_cache_test.go new file mode 100644 index 00000000..6af99666 --- /dev/null +++ b/pkg/mcp/orientation_cache_test.go @@ -0,0 +1,98 @@ +package mcp_test + +import ( + "context" + "testing" + "time" + + sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/stretchr/testify/require" + + "github.com/jlrickert/tapper/pkg/mcp" + "github.com/jlrickert/tapper/pkg/tapper" +) + +func newCachedFlightSession(t *testing.T, backend *perCallFlightBackend, ttl time.Duration) (*sdkmcp.ClientSession, context.Context, func(time.Duration)) { + t.Helper() + ctx := context.Background() + sb := newTestSandbox(t) + tap, err := tapper.NewTap(tapper.TapOptions{Runtime: sb.Runtime()}) + require.NoError(t, err) + server := mcp.NewServer(tap, "test", mcp.KegDefaults{}, mcp.ServerOptions{ + OrientationProvider: backend, FlightProvider: backend, KegProvider: backend, + KegSearchProvider: backend, IdentityProvider: backend, + OrientationCacheTTL: ttl, + }) + return connectFlightSession(t, ctx, server, nil), ctx, func(d time.Duration) { sb.Advance(d) } +} + +func (p *perCallFlightBackend) resolveCount() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.resolves +} + +func callKegList(t *testing.T, ctx context.Context, session *sdkmcp.ClientSession) *sdkmcp.CallToolResult { + t.Helper() + result, err := session.CallTool(ctx, &sdkmcp.CallToolParams{Name: "keg_list", Arguments: map[string]any{}}) + require.NoError(t, err) + return result +} + +// A burst of tool calls reuses one resolved orientation; it is resolved again +// once the window passes. +func TestOrientationCacheReusesResolutionWithinWindow(t *testing.T) { + backend := newPerCallFlightBackend() + session, ctx, advance := newCachedFlightSession(t, backend, 10*time.Second) + + require.False(t, callKegList(t, ctx, session).IsError) + before := backend.resolveCount() + for range 5 { + require.False(t, callKegList(t, ctx, session).IsError) + } + require.Equal(t, before, backend.resolveCount(), "calls inside the window must not resolve again") + + advance(11 * time.Second) + require.False(t, callKegList(t, ctx, session).IsError) + require.Equal(t, before+1, backend.resolveCount(), "an expired view must be resolved again") +} + +// orient always answers with current authority, and the calls after it see +// what it saw. +func TestOrientationCacheNeverServesOrient(t *testing.T) { + backend := newPerCallFlightBackend() + session, ctx, _ := newCachedFlightSession(t, backend, time.Hour) + + require.False(t, callKegList(t, ctx, session).IsError) + before := backend.resolveCount() + oriented, err := session.CallTool(ctx, &sdkmcp.CallToolParams{Name: "orient", Arguments: map[string]any{}}) + require.NoError(t, err) + require.False(t, oriented.IsError, extractText(t, oriented)) + require.False(t, callKegList(t, ctx, session).IsError) + require.Greater(t, backend.resolveCount(), before, "orient must discard the cached view") +} + +// A failed call may mean authority changed, so the next call resolves again. +func TestOrientationCacheDropsViewAfterFailure(t *testing.T) { + backend := newPerCallFlightBackend() + session, ctx, _ := newCachedFlightSession(t, backend, time.Hour) + + require.False(t, callKegList(t, ctx, session).IsError) + failed := callCatKeg(t, ctx, session, "@elsewhere/uncovered") + require.True(t, failed.IsError, extractText(t, failed)) + before := backend.resolveCount() + require.False(t, callKegList(t, ctx, session).IsError) + require.Equal(t, before+1, backend.resolveCount()) +} + +// Without a window every call resolves current authority, which is what a +// server resolving in process (a Hub's own endpoint) wants. +func TestOrientationCacheIsOffByDefault(t *testing.T) { + backend := newPerCallFlightBackend() + session, ctx, _ := newCachedFlightSession(t, backend, 0) + + require.False(t, callKegList(t, ctx, session).IsError) + before := backend.resolveCount() + require.False(t, callKegList(t, ctx, session).IsError) + require.Equal(t, before+1, backend.resolveCount()) +} diff --git a/pkg/mcp/server.go b/pkg/mcp/server.go index 0a409e31..b95bb57a 100644 --- a/pkg/mcp/server.go +++ b/pkg/mcp/server.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "strings" + "time" "github.com/jlrickert/tapper/pkg/apicontract" sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp" @@ -47,6 +48,13 @@ type ServerOptions struct { // attachment transfer tools: see registerFileTools. The zero value is the // safe one, so a caller that forgets it gets the hosted surface. SharedFilesystem bool + // OrientationCacheTTL, when positive, lets a session reuse its resolved + // orientation for that long instead of resolving it on every tool call. + // It suits a client of a remote Hub, where each resolution costs several + // requests and the Hub rechecks authority on every KEG request anyway. + // orient, session_refresh, flight mutations, and any failed call always + // discard the cached view. + OrientationCacheTTL time.Duration } // NewServer builds an MCP server with all registered tools. @@ -78,6 +86,10 @@ func NewServer(tap *tapper.Tap, version string, defaults KegDefaults, opts ...Se } defaults.gate = newSessionFlightGate(opt.OrientationProvider) defaults.gate.clientVersion = version + if opt.OrientationCacheTTL > 0 && tap != nil && tap.Runtime != nil { + defaults.gate.cacheTTL = opt.OrientationCacheTTL + defaults.gate.now = tap.Runtime.Clock().Now + } defaults.gate.logger = opt.Logger if defaults.gate.logger == nil && tap != nil && tap.Runtime != nil { defaults.gate.logger = tap.Runtime.Logger() diff --git a/pkg/mcp/session_flight.go b/pkg/mcp/session_flight.go index 087f986f..9a21391b 100644 --- a/pkg/mcp/session_flight.go +++ b/pkg/mcp/session_flight.go @@ -9,6 +9,7 @@ import ( "slices" "strings" "sync" + "time" "github.com/jlrickert/tapper/pkg/apicontract" sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp" @@ -144,15 +145,30 @@ type sessionFlightGate struct { clientVersion string logger *slog.Logger - mu sync.Mutex - states map[string]*flightSessionState - srv *sdkmcp.Server + // cacheTTL, when positive, reuses a session's resolved orientation for + // that long instead of resolving it again on every tool call. now is the + // clock it is measured against. + cacheTTL time.Duration + now func() time.Time + + mu sync.Mutex + states map[string]*flightSessionState + resolved map[string]cachedOrientation + srv *sdkmcp.Server +} + +// cachedOrientation is one session's most recent per-call resolution. +type cachedOrientation struct { + root, selected string + orientation *Orientation + at time.Time } func newSessionFlightGate(provider OrientationProvider) *sessionFlightGate { g := &sessionFlightGate{ provider: provider, states: map[string]*flightSessionState{}, + resolved: map[string]cachedOrientation{}, } return g } @@ -283,13 +299,52 @@ func (g *sessionFlightGate) resolveCall(ctx context.Context, sessionID, selected if pinned.root != nil { rootRef = pinned.root.Name } + if candidate, ok := g.cachedResolution(sessionID, rootRef, selected); ok { + return makeOrientationContext(candidate) + } candidate, err := resolver.Resolve(ctx, rootRef, selected) if err != nil { + g.forgetResolution(sessionID) return nil, err } + g.rememberResolution(sessionID, rootRef, selected, candidate) return makeOrientationContext(candidate) } +// cachedResolution returns a session's resolved orientation while it is +// fresh. Reuse is safe because the Hub rechecks authority on every KEG +// request; the cache only spares the client from rebuilding its view of that +// authority on every tool call. +func (g *sessionFlightGate) cachedResolution(sessionID, root, selected string) (*Orientation, bool) { + if g.cacheTTL <= 0 || g.now == nil { + return nil, false + } + g.mu.Lock() + defer g.mu.Unlock() + entry, ok := g.resolved[sessionID] + if !ok || entry.root != root || entry.selected != selected || g.now().Sub(entry.at) >= g.cacheTTL { + return nil, false + } + return entry.orientation, true +} + +func (g *sessionFlightGate) rememberResolution(sessionID, root, selected string, orientation *Orientation) { + if g.cacheTTL <= 0 || g.now == nil || orientation == nil { + return + } + g.mu.Lock() + defer g.mu.Unlock() + g.resolved[sessionID] = cachedOrientation{root: root, selected: selected, orientation: orientation, at: g.now()} +} + +// forgetResolution drops a session's cached orientation so the next call +// resolves current authority. +func (g *sessionFlightGate) forgetResolution(sessionID string) { + g.mu.Lock() + defer g.mu.Unlock() + delete(g.resolved, sessionID) +} + func makeOrientationContext(candidate *Orientation) (*orientationContext, error) { if candidate == nil { candidate = &Orientation{} @@ -628,6 +683,11 @@ func (g *sessionFlightGate) middleware(next sdkmcp.MethodHandler) sdkmcp.MethodH } if method == "tools/call" { params, _ := req.GetParams().(*sdkmcp.CallToolParamsRaw) + if params != nil && (params.Name == "orient" || params.Name == "session_refresh" || isFlightMutationTool(params.Name)) { + // These ask for, or change, current authority: never answer + // them or the calls after them from a cached resolution. + g.forgetResolution(sessionID) + } if allowed := allowedTools(current); params != nil && allowed != nil && !allowed[params.Name] { return errorResult(lockedError(current)), nil } @@ -686,7 +746,12 @@ func (g *sessionFlightGate) middleware(next sdkmcp.MethodHandler) sdkmcp.MethodH return orientationFailureResult(err), nil } } - return next(ctx, method, req) + result, err := next(ctx, method, req) + if called, ok := result.(*sdkmcp.CallToolResult); err != nil || (ok && called.IsError) { + // A failure may mean authority changed under the cached view. + g.forgetResolution(sessionID) + } + return result, err } if method == "resources/read" || method == "resources/subscribe" { if params, ok := req.GetParams().(*sdkmcp.ReadResourceParams); ok && params.URI == orientResourceURI { diff --git a/pkg/mcp/tools_read.go b/pkg/mcp/tools_read.go index c0aa527c..143b5731 100644 --- a/pkg/mcp/tools_read.go +++ b/pkg/mcp/tools_read.go @@ -98,7 +98,7 @@ func registerCat(srv *sdkmcp.Server, tap *tapper.Tap, defaults KegDefaults) { type listInput struct { Query string `json:"query,omitempty" jsonschema:"boolean query expression to filter nodes. Supports tags ('golang'), key=value attributes ('entity=plan'), and dot-prefix stats fields ('.created>2026-01-01', '.accessCount>=5', '.hash=abc123'). Combine with 'and', 'or', 'not'."` Keg string `json:"keg,omitempty" jsonschema:"keg alias (uses default if empty)"` - Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates read one file per node."` + Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates are resolved by the Hub for the returned page in one batch."` IdOnly bool `json:"id_only,omitempty" jsonschema:"return node IDs only"` Reverse bool `json:"reverse,omitempty" jsonschema:"reverse output order"` Sort string `json:"sort,omitempty" jsonschema:"sort order: 'id' (default), 'updated', 'created', or 'accessed'"` @@ -136,9 +136,9 @@ func registerList(srv *sdkmcp.Server, tap *tapper.Tap, defaults KegDefaults) { // --- node_search --- type grepInput struct { - Query string `json:"query" jsonschema:"regex pattern to search node content"` + Query string `json:"query" jsonschema:"regex pattern to search node content, matched per line. Local KEGs use Go RE2 syntax; Hub KEGs use PostgreSQL regex, where \\y (not \\b) is a word boundary and lookahead is allowed"` Keg string `json:"keg,omitempty" jsonschema:"keg alias (uses default if empty)"` - Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates read one file per node. Use id_only for compact MCP output."` + Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates are resolved by the Hub for the returned page in one batch. Use id_only for compact MCP output."` IdOnly bool `json:"id_only,omitempty" jsonschema:"return node IDs only (recommended for MCP to reduce token usage)"` Reverse bool `json:"reverse,omitempty" jsonschema:"reverse output order"` IgnoreCase bool `json:"ignore_case,omitempty" jsonschema:"case-insensitive matching"` @@ -180,7 +180,7 @@ func registerGrep(srv *sdkmcp.Server, tap *tapper.Tap, defaults KegDefaults) { type tagsInput struct { Query string `json:"query,omitempty" jsonschema:"boolean expression to filter by tags, attributes, and dot-prefix stats fields (e.g. '.created>2026-01-01 and entity=plan')"` Keg string `json:"keg,omitempty" jsonschema:"keg alias (uses default if empty)"` - Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates read one file per node."` + Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates are resolved by the Hub for the returned page in one batch."` IdOnly bool `json:"id_only,omitempty" jsonschema:"return node IDs only"` Reverse bool `json:"reverse,omitempty" jsonschema:"reverse output order"` Limit int `json:"limit,omitempty" jsonschema:"maximum results to return (default: 50, 0 in request means use default, -1 for unlimited)"` @@ -218,7 +218,7 @@ func registerTags(srv *sdkmcp.Server, tap *tapper.Tap, defaults KegDefaults) { type backlinksInput struct { NodeIDs []string `json:"node_ids" jsonschema:"target node IDs to find incoming links for (results merged and deduplicated)"` Keg string `json:"keg,omitempty" jsonschema:"keg alias (uses default if empty)"` - Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates read one file per node."` + Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates are resolved by the Hub for the returned page in one batch."` IdOnly bool `json:"id_only,omitempty" jsonschema:"return node IDs only"` Reverse bool `json:"reverse,omitempty" jsonschema:"reverse output order"` Limit int `json:"limit,omitempty" jsonschema:"maximum results to return (default: 50, 0 in request means use default, -1 for unlimited)"` @@ -256,7 +256,7 @@ func registerBacklinks(srv *sdkmcp.Server, tap *tapper.Tap, defaults KegDefaults type linksInput struct { NodeIDs []string `json:"node_ids" jsonschema:"source node IDs to find outgoing links for (results merged and deduplicated)"` Keg string `json:"keg,omitempty" jsonschema:"keg alias (uses default if empty)"` - Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates read one file per node."` + Format string `json:"format,omitempty" jsonschema:"output format template. Legacy verbs %i (id), %t (title), %d (updated), %c (created), %a (accessed); %% is a literal percent. Named selectors use %{...}: a bare word names a metadata key such as %{type} or %{status}, a leading dot names a statistics field such as %{.accessCount} or %{.omega}, and %{tags} is the node's tag list. Backslash escapes are interpreted: \\t tab, \\n newline, \\r return, \\\\ backslash. Selectors other than id, title, and the three dates are resolved by the Hub for the returned page in one batch."` IdOnly bool `json:"id_only,omitempty" jsonschema:"return node IDs only"` Reverse bool `json:"reverse,omitempty" jsonschema:"reverse output order"` Limit int `json:"limit,omitempty" jsonschema:"maximum results to return (default: 50, 0 in request means use default, -1 for unlimited)"` diff --git a/pkg/tapapi/tapapi.go b/pkg/tapapi/tapapi.go new file mode 100644 index 00000000..6ce10fae --- /dev/null +++ b/pkg/tapapi/tapapi.go @@ -0,0 +1,85 @@ +// Package tapapi defines the wire contract for complete TAP operations a Hub +// executes near the data, under /api/v1/tap/*. +// +// A per-KEG route performs one KEG operation. A TAP operation is a whole +// client-facing workflow (a relationship lookup plus the fields its listing +// renders, a tag listing, a multi-KEG read) that would otherwise take several +// round trips. The Hub resolves every KEG a request names through its KEG +// authority boundary; nothing here grants access. +// +// Rendering stays on the client: responses carry structured rows and resolved +// field values, never formatted lines. +package tapapi + +import "github.com/jlrickert/tapper/pkg/keg" + +// Paths of the TAP operations, relative to the Hub's /api/v1 root. +const ( + PathLinks = "/tap/links" + PathBacklinks = "/tap/backlinks" + PathTags = "/tap/tags" + PathCat = "/tap/cat" +) + +// RelatedRequest asks for a node set's links or backlinks in one KEG, paged +// and with the named field selectors resolved for every returned row. +type RelatedRequest struct { + // Keg is the qualified KEG reference, "@namespace/alias". + Keg string `json:"keg"` + NodeIDs []string `json:"node_ids"` + Fields []string `json:"fields,omitempty"` + Offset int `json:"offset,omitempty"` + Limit int `json:"limit,omitempty"` +} + +// RelatedResponse is one page of related nodes. Pairs attributes every +// returned row to the input node that produced it; Total counts related +// nodes before paging. +type RelatedResponse struct { + Rows []keg.ListViewRow `json:"rows"` + Pairs []keg.RelatedPair `json:"pairs"` + Total int `json:"total"` +} + +// TagsRequest lists tags, or with a Query, the nodes it selects rendered as +// rows with the named fields resolved. +type TagsRequest struct { + Keg string `json:"keg"` + Query string `json:"query,omitempty"` + Fields []string `json:"fields,omitempty"` + Offset int `json:"offset,omitempty"` + Limit int `json:"limit,omitempty"` +} + +// TagsResponse carries Tags when the request had no query, otherwise Rows. +// Total counts before paging. +type TagsResponse struct { + Tags []string `json:"tags,omitempty"` + Rows []keg.ListViewRow `json:"rows,omitempty"` + Total int `json:"total"` +} + +// CatRequest reads nodes across any number of KEGs on this Hub in one call. +// Each reference is qualified, "keg:@namespace/alias/id". Touch records the +// read as an open. +type CatRequest struct { + Nodes []string `json:"nodes"` + Touch bool `json:"touch,omitempty"` +} + +// CatNode is one requested node, in request order. A node that cannot be read +// carries Error and Code instead of content, so one failure does not hide the +// rest of the batch. +type CatNode struct { + Ref string `json:"ref"` + Content string `json:"content,omitempty"` + Meta string `json:"meta,omitempty"` + Hash string `json:"hash,omitempty"` + Error string `json:"error,omitempty"` + Code string `json:"code,omitempty"` +} + +// CatResponse returns the requested nodes in request order. +type CatResponse struct { + Nodes []CatNode `json:"nodes"` +} diff --git a/pkg/tapper/invocation_telemetry_test.go b/pkg/tapper/invocation_telemetry_test.go index 7de35964..cd1ff638 100644 --- a/pkg/tapper/invocation_telemetry_test.go +++ b/pkg/tapper/invocation_telemetry_test.go @@ -19,7 +19,7 @@ type telemetryRoundTripFunc func(*http.Request) (*http.Response, error) func (f telemetryRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { if req.URL.Path == "/api/version" { - return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"server_version":"test","api_versions":["2026-09-11"]}`))}, nil + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"server_version":"test","api_versions":["2026-09-28"]}`))}, nil } return f(req) diff --git a/pkg/tapper/operations.go b/pkg/tapper/operations.go new file mode 100644 index 00000000..6ea0e11c --- /dev/null +++ b/pkg/tapper/operations.go @@ -0,0 +1,184 @@ +package tapper + +import ( + "context" + "fmt" + "slices" + "sort" + "strings" + + "github.com/jlrickert/tapper/pkg/keg" + "github.com/jlrickert/tapper/pkg/tapapi" +) + +// Operations executes complete TAP operations against a resolved KEG: the +// relationship lookup plus the fields its listing renders, or a tag listing, +// as one unit. It is the seam that lets a Hub-backed KEG answer the whole +// operation next to the data instead of the client composing KEG calls. +// +// The default forwards to the KEG's Hub over /api/v1/tap/* when the KEG is a +// Hub KEG and composes KEG operations locally otherwise. A Hub serving its +// own MCP endpoint injects an implementation backed by its TAP service. +type Operations interface { + // Related returns one page of a node set's links or backlinks in k. + Related(ctx context.Context, k keg.Keg, direction keg.RelatedDirection, in tapapi.RelatedRequest) (*tapapi.RelatedResponse, error) + // Tags lists k's tags, or with a query, one page of the nodes it selects. + Tags(ctx context.Context, k keg.Keg, in tapapi.TagsRequest) (*tapapi.TagsResponse, error) +} + +// operations returns the injected Operations or the default. +func (t *Tap) operations() Operations { + if t.Operations != nil { + return t.Operations + } + return DefaultOperations{} +} + +// hubKeg is a KEG served by a Tapper Hub that can carry a complete TAP +// operation to it. +type hubKeg interface { + HubPost(ctx context.Context, path, op string, in, out any) error + Target() *keg.Target +} + +// DefaultOperations forwards to the KEG's Hub when it is a Hub KEG and +// composes KEG operations otherwise. +type DefaultOperations struct{} + +func (DefaultOperations) Related(ctx context.Context, k keg.Keg, direction keg.RelatedDirection, in tapapi.RelatedRequest) (*tapapi.RelatedResponse, error) { + if hub, ref, ok := hubKegRef(k); ok { + in.Keg = ref + path := tapapi.PathLinks + if direction == keg.RelatedBacklinks { + path = tapapi.PathBacklinks + } + var out tapapi.RelatedResponse + if err := hub.HubPost(ctx, path, "Related", in, &out); err != nil { + return nil, err + } + return &out, nil + } + return LocalOperations{}.Related(ctx, k, direction, in) +} + +func (DefaultOperations) Tags(ctx context.Context, k keg.Keg, in tapapi.TagsRequest) (*tapapi.TagsResponse, error) { + if hub, ref, ok := hubKegRef(k); ok { + in.Keg = ref + var out tapapi.TagsResponse + if err := hub.HubPost(ctx, tapapi.PathTags, "Tags", in, &out); err != nil { + return nil, err + } + return &out, nil + } + return LocalOperations{}.Tags(ctx, k, in) +} + +func hubKegRef(k keg.Keg) (hubKeg, string, bool) { + hub, ok := k.(hubKeg) + if !ok { + return nil, "", false + } + target := hub.Target() + if target == nil || target.Namespace == "" || target.KegName == "" { + return nil, "", false + } + return hub, "@" + target.Namespace + "/" + target.KegName, true +} + +// LocalOperations composes TAP operations from KEG operations. Field values +// for a page are resolved with one ListView call, never per node. +type LocalOperations struct{} + +func (LocalOperations) Related(ctx context.Context, k keg.Keg, direction keg.RelatedDirection, in tapapi.RelatedRequest) (*tapapi.RelatedResponse, error) { + if in.Offset < 0 || in.Limit < 0 { + return nil, fmt.Errorf("offset and limit must be >= 0: %w", keg.ErrInvalid) + } + ids := make([]keg.NodeId, 0, len(in.NodeIDs)) + for _, raw := range in.NodeIDs { + id, err := keg.ParseNode(raw) + if err != nil || id == nil { + return nil, fmt.Errorf("invalid node id %q: %w", raw, keg.ErrInvalid) + } + ids = append(ids, *id) + } + related, err := k.RelatedNodes(ctx, keg.RelatedNodesOptions{NodeIDs: ids, Direction: direction}) + if err != nil { + return nil, err + } + entries := pageOf(related.Entries, in.Offset, in.Limit) + rows, err := projectRows(ctx, k, entries, in.Fields) + if err != nil { + return nil, err + } + onPage := make(map[string]bool, len(entries)) + for _, entry := range entries { + onPage[entry.ID] = true + } + pairs := make([]keg.RelatedPair, 0, len(related.Pairs)) + for _, pair := range related.Pairs { + if onPage[pair.To] { + pairs = append(pairs, pair) + } + } + return &tapapi.RelatedResponse{Rows: rows, Pairs: pairs, Total: len(related.Entries)}, nil +} + +func (LocalOperations) Tags(ctx context.Context, k keg.Keg, in tapapi.TagsRequest) (*tapapi.TagsResponse, error) { + if in.Offset < 0 || in.Limit < 0 { + return nil, fmt.Errorf("offset and limit must be >= 0: %w", keg.ErrInvalid) + } + if strings.TrimSpace(in.Query) == "" { + listing, err := k.ListEntries(ctx, keg.ListEntriesOptions{}) + if err != nil { + return nil, err + } + tags := slices.Clone(listing.Tags) + sort.Strings(tags) + return &tapapi.TagsResponse{Tags: pageOf(tags, in.Offset, in.Limit), Total: len(tags)}, nil + } + view, err := k.ListView(ctx, keg.ListViewOptions{Query: in.Query, Fields: in.Fields, Offset: in.Offset, Limit: in.Limit}) + if err != nil { + return nil, err + } + return &tapapi.TagsResponse{Rows: view.Rows, Total: view.TotalMatches}, nil +} + +// projectRows resolves fields for a known page of entries in one ListView. +func projectRows(ctx context.Context, k keg.Keg, entries []keg.NodeIndexEntry, fields []string) ([]keg.ListViewRow, error) { + rows := make([]keg.ListViewRow, len(entries)) + for i, entry := range entries { + rows[i] = keg.ListViewRow{Entry: entry} + } + if len(fields) == 0 || len(entries) == 0 { + return rows, nil + } + ids := make([]keg.NodeId, 0, len(entries)) + for _, entry := range entries { + if id, err := keg.ParseNode(entry.ID); err == nil && id != nil { + ids = append(ids, *id) + } + } + view, err := k.ListView(ctx, keg.ListViewOptions{NodeIDs: ids, Fields: fields}) + if err != nil { + return nil, fmt.Errorf("unable to resolve listing fields: %w", err) + } + resolved := make(map[string]map[string]string, len(view.Rows)) + for _, row := range view.Rows { + resolved[row.Entry.ID] = row.Fields + } + for i := range rows { + rows[i].Fields = resolved[rows[i].Entry.ID] + } + return rows, nil +} + +func pageOf[T any](items []T, offset, limit int) []T { + if offset >= len(items) { + return []T{} + } + items = items[offset:] + if limit > 0 && len(items) > limit { + items = items[:limit] + } + return items +} diff --git a/pkg/tapper/relationship_observer_test.go b/pkg/tapper/relationship_observer_test.go index 71ef1968..4a82b1f0 100644 --- a/pkg/tapper/relationship_observer_test.go +++ b/pkg/tapper/relationship_observer_test.go @@ -10,14 +10,15 @@ import ( type observedLinksKeg struct{ keg.Keg } -func (k *observedLinksKeg) RelatedNodes(_ context.Context, o keg.RelatedNodesOptions) ([]keg.NodeIndexEntry, error) { - if len(o.NodeIDs) > 1 { - return []keg.NodeIndexEntry{{ID: "3"}, {ID: "4"}}, nil +// Node 1 relates to 3 and node 2 relates to 4. +func (k *observedLinksKeg) RelatedNodes(_ context.Context, o keg.RelatedNodesOptions) (*keg.RelatedNodesResult, error) { + out := &keg.RelatedNodesResult{} + for _, id := range o.NodeIDs { + to := map[int]string{1: "3", 2: "4"}[id.ID] + out.Entries = append(out.Entries, keg.NodeIndexEntry{ID: to}) + out.Pairs = append(out.Pairs, keg.RelatedPair{From: id.Path(), To: to}) } - if o.NodeIDs[0].ID == 1 { - return []keg.NodeIndexEntry{{ID: "3"}}, nil - } - return []keg.NodeIndexEntry{{ID: "4"}}, nil + return out, nil } func TestRelationshipObserverOnlyActualPairsInResultPage(t *testing.T) { k := &observedLinksKeg{} @@ -55,14 +56,25 @@ type crossObservedKeg struct { malformed bool } -func (k *crossObservedKeg) RelatedNodes(_ context.Context, o keg.RelatedNodesOptions) ([]keg.NodeIndexEntry, error) { - if k.fail && len(o.NodeIDs) == 1 { +// Every input relates to the same two cross-keg nodes. +func (k *crossObservedKeg) RelatedNodes(_ context.Context, o keg.RelatedNodesOptions) (*keg.RelatedNodesResult, error) { + if k.fail { return nil, fmt.Errorf("lookup failed") } + targets := []string{"keg:other/4", "keg:@team/third/4"} if k.malformed { - return []keg.NodeIndexEntry{{ID: "keg:@broken"}}, nil + targets = []string{"keg:@broken"} + } + out := &keg.RelatedNodesResult{} + for _, target := range targets { + out.Entries = append(out.Entries, keg.NodeIndexEntry{ID: target}) + } + for _, id := range o.NodeIDs { + for _, target := range targets { + out.Pairs = append(out.Pairs, keg.RelatedPair{From: id.Path(), To: target}) + } } - return []keg.NodeIndexEntry{{ID: "keg:other/4"}, {ID: "keg:@team/third/4"}, {ID: "keg:other/4"}}, nil + return out, nil } func TestRelationshipObserverFullIdentitiesAndFailures(t *testing.T) { for _, back := range []bool{false, true} { diff --git a/pkg/tapper/tap.go b/pkg/tapper/tap.go index ef54dc9e..483eb797 100644 --- a/pkg/tapper/tap.go +++ b/pkg/tapper/tap.go @@ -41,6 +41,11 @@ type Tap struct { // the standard config-driven resolution. KegResolver func(ctx context.Context, opts KegTargetOptions, role FlightRole) (keg.Keg, error) + // Operations, when non-nil, executes complete TAP operations (links, + // backlinks, tags) instead of DefaultOperations. A Hub serving its own MCP + // endpoint injects its in-process TAP service here. + Operations Operations + // Version is the version of the binary running this Tap, already // normalized by NewTap. `tap integrate` stamps it into every plugin // manifest it extracts, so an installed plugin always reports the tap diff --git a/pkg/tapper/tap_cat.go b/pkg/tapper/tap_cat.go index 3808c225..4ae06dbb 100644 --- a/pkg/tapper/tap_cat.go +++ b/pkg/tapper/tap_cat.go @@ -242,19 +242,20 @@ func (t *Tap) catSingleNode(ctx context.Context, k keg.Keg, nodeID string, opts return "", err } - view, err := k.ReadNode(ctx, node) + // One touched read: the node and its access update travel together. + views, err := k.ReadNodes(ctx, keg.ReadNodesOptions{NodeIDs: []keg.NodeId{node}, Touch: true}) + if err == nil && len(views) != 1 { + err = keg.ErrNotExist + } if err != nil { if errors.Is(err, keg.ErrNotExist) { return "", fmt.Errorf("node %s not found in %s: %w", node.Path(), describeKeg(k), err) } return "", fmt.Errorf("unable to read node %s in %s: %w", node.Path(), describeKeg(k), err) } + view := views[0] content, meta := view.Content, view.Meta - if err := k.Touch(ctx, node); err != nil { - return "", fmt.Errorf("unable to update node access: %w", err) - } - if opts.ContentOnly { return string(content), nil } @@ -336,19 +337,20 @@ func (t *Tap) catSingleNodeForStream(ctx context.Context, k keg.Keg, nodeID stri return "", err } - view, err := k.ReadNode(ctx, node) + // One touched read: the node and its access update travel together. + views, err := k.ReadNodes(ctx, keg.ReadNodesOptions{NodeIDs: []keg.NodeId{node}, Touch: true}) + if err == nil && len(views) != 1 { + err = keg.ErrNotExist + } if err != nil { if errors.Is(err, keg.ErrNotExist) { return "", fmt.Errorf("node %s not found in %s: %w", node.Path(), describeKeg(k), err) } return "", fmt.Errorf("unable to read node %s in %s: %w", node.Path(), describeKeg(k), err) } + view := views[0] content, meta := view.Content, view.Meta - if err := k.Touch(ctx, node); err != nil { - return "", fmt.Errorf("unable to update node access: %w", err) - } - id := node.Path() if opts.ContentOnly { diff --git a/pkg/tapper/tap_list.go b/pkg/tapper/tap_list.go index 2bf0ddf4..541f7668 100644 --- a/pkg/tapper/tap_list.go +++ b/pkg/tapper/tap_list.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "github.com/jlrickert/tapper/pkg/tapapi" "regexp" "strings" "time" @@ -35,7 +36,8 @@ type ListOptions struct { // renders a literal percent. Named selectors use %{...}: a bare word // names a metadata key (%{type}), a leading dot names a statistics field // (%{.accessCount}), and %{tags} is the tag list. Selectors other than - // id, title, and the three dates cost one read per node. + // id, title, and the three dates are resolved by the keg for the + // returned page in one batch. Format string IdOnly bool @@ -64,7 +66,8 @@ type BacklinksOptions struct { // renders a literal percent. Named selectors use %{...}: a bare word // names a metadata key (%{type}), a leading dot names a statistics field // (%{.accessCount}), and %{tags} is the tag list. Selectors other than - // id, title, and the three dates cost one read per node. + // id, title, and the three dates are resolved by the keg for the + // returned page in one batch. Format string IdOnly bool @@ -90,7 +93,8 @@ type LinksOptions struct { // renders a literal percent. Named selectors use %{...}: a bare word // names a metadata key (%{type}), a leading dot names a statistics field // (%{.accessCount}), and %{tags} is the tag list. Selectors other than - // id, title, and the three dates cost one read per node. + // id, title, and the three dates are resolved by the keg for the + // returned page in one batch. Format string IdOnly bool @@ -115,7 +119,8 @@ type GrepOptions struct { // renders a literal percent. Named selectors use %{...}: a bare word // names a metadata key (%{type}), a leading dot names a statistics field // (%{.accessCount}), and %{tags} is the tag list. Selectors other than - // id, title, and the three dates cost one read per node. + // id, title, and the three dates are resolved by the keg for the + // returned page in one batch. Format string IdOnly bool @@ -150,7 +155,8 @@ type TagsOptions struct { // renders a literal percent. Named selectors use %{...}: a bare word // names a metadata key (%{type}), a leading dot names a statistics field // (%{.accessCount}), and %{tags} is the tag list. Selectors other than - // id, title, and the three dates cost one read per node. + // id, title, and the three dates are resolved by the keg for the + // returned page in one batch. Format string IdOnly bool @@ -200,19 +206,15 @@ func (t *Tap) List(ctx context.Context, opts ListOptions) ([]string, error) { }) switch { case err == nil: - t.warnStaleIndex(view.IndexedCount, view.NodeCount) - return renderListView(compiled, view.Rows, renderOptions{ - Format: opts.Format, IdOnly: opts.IdOnly, Reverse: opts.Reverse, - }), nil - case errors.Is(err, keg.ErrListViewUnsupported): - // Hub predates the endpoint; assemble the listing here instead. case strings.TrimSpace(opts.Query) != "": return []string{}, fmt.Errorf("invalid query expression: %w", err) default: return []string{}, fmt.Errorf("unable to list keg: %w", err) } - - return t.listClientSide(ctx, k, opts, compiled) + t.warnStaleIndex(view.IndexedCount, view.NodeCount) + return renderListView(compiled, view.Rows, renderOptions{ + Format: opts.Format, IdOnly: opts.IdOnly, Reverse: opts.Reverse, + }), nil } // resolveListFormat picks the format for a listing: an explicit --format wins, @@ -278,47 +280,6 @@ func (t *Tap) warnStaleIndex(indexed, total int) { } } -// listClientSide reproduces the listing locally for hubs that predate the -// server-resolved endpoint. It is strictly slower — field values cost a read -// per row — so it exists only to keep older deployments working. -func (t *Tap) listClientSide(ctx context.Context, k keg.Keg, opts ListOptions, compiled compiledFormat) ([]string, error) { - listing, err := k.ListEntries(ctx, keg.ListEntriesOptions{Query: opts.Query}) - if err != nil { - if strings.TrimSpace(opts.Query) != "" { - return []string{}, fmt.Errorf("invalid query expression: %w", err) - } - return []string{}, fmt.Errorf("unable to list keg: %w", err) - } - - entries := listing.Entries - t.warnStaleIndex(listing.IndexedCount, listing.NodeCount) - - if strings.TrimSpace(opts.Query) != "" { - sortNodeIndexEntries(entries) - } - - switch opts.Sort { - case SortByDefault, SortByID: - // already sorted by ID from dex.Nodes() / sortNodeIndexEntries - case SortByUpdated: - sortNodeIndexEntriesByTime(entries, func(e keg.NodeIndexEntry) time.Time { return e.Updated }) - case SortByCreated: - sortNodeIndexEntriesByTime(entries, func(e keg.NodeIndexEntry) time.Time { return e.Created }) - case SortByAccessed: - sortNodeIndexEntriesByTime(entries, func(e keg.NodeIndexEntry) time.Time { return e.Accessed }) - } - - entries = applyOffset(entries, opts.Offset) - - if opts.Limit > 0 && len(entries) > opts.Limit { - entries = entries[:opts.Limit] - } - - return t.renderNodeEntries(ctx, k, entries, renderOptions{ - Format: opts.Format, IdOnly: opts.IdOnly, Reverse: opts.Reverse, - }) -} - func (t *Tap) Backlinks(ctx context.Context, opts BacklinksOptions) ([]string, error) { if opts.Offset < 0 { return []string{}, fmt.Errorf("offset must be >= 0, got %d", opts.Offset) @@ -367,9 +328,9 @@ type relatedListOptions struct { Direction keg.RelatedDirection } -// resolveAndLookupLinks is a shared helper for Backlinks and Links. It resolves -// the keg, validates the node IDs, calls the provided lookup function against the -// dex for each node, merges and deduplicates results, and renders the entries. +// resolveAndLookupLinks is the shared body of Backlinks and Links. The whole +// operation — relationship lookup, paging, and the fields the format renders — +// runs through Operations, so a Hub KEG answers it in one request. func (t *Tap) resolveAndLookupLinks(ctx context.Context, opts relatedListOptions) ([]string, error) { if len(opts.NodeIDs) == 0 { return []string{}, fmt.Errorf("at least one node ID is required") @@ -379,77 +340,61 @@ func (t *Tap) resolveAndLookupLinks(ctx context.Context, opts relatedListOptions if err != nil { return []string{}, fmt.Errorf("unable to open keg: %w", err) } - ids := make([]keg.NodeId, 0, len(opts.NodeIDs)) + ids := make([]string, 0, len(opts.NodeIDs)) for _, nodeID := range opts.NodeIDs { // Intentionally NOT routed through resolveNodeArg: a cross-keg ref would - // produce related nodes owned by a different keg, but the dedup key - // (rel.Path()) and the entry rendering below (dex.GetRef) both assume a - // single owning dex. Carrying each related node's home keg through dedup, - // sort, offset/limit, and render is a larger change than this slice - // covers, so links/backlinks stay scoped to the current keg. + // produce related nodes owned by a different keg, but paging and + // rendering assume a single owning keg, so links/backlinks stay scoped + // to the current keg. id, err := parseNodeID(nodeID) if err != nil { return []string{}, err } - - ids = append(ids, id) + ids = append(ids, id.Path()) } - entries, err := k.RelatedNodes(ctx, keg.RelatedNodesOptions{NodeIDs: ids, Direction: opts.Direction}) + compiled, err := compileListFormat(opts.Render.Format) + if err != nil { + return nil, err + } + related, err := t.operations().Related(ctx, k, opts.Direction, tapapi.RelatedRequest{ + NodeIDs: ids, + Fields: compiled.selectorTexts(opts.Render.IdOnly), + Offset: opts.Offset, + Limit: opts.Limit, + }) if err != nil { if strings.Contains(err.Error(), "keg not initialized") { return []string{}, err } if errors.Is(err, keg.ErrNotExist) { - return []string{}, fmt.Errorf("node %s not found in %s", ids[0].Path(), describeKeg(k)) + return []string{}, fmt.Errorf("node %s not found in %s", ids[0], describeKeg(k)) } return []string{}, err } - - entries = applyOffset(entries, opts.Offset) - - if opts.Limit > 0 && len(entries) > opts.Limit { - entries = entries[:opts.Limit] - } - - rendered, err := t.renderNodeEntries(ctx, k, entries, opts.Render) - if err != nil { - return nil, err - } + rendered := renderListView(compiled, related.Rows, opts.Render) if observer, ok := ctx.Value(relationshipObserverKey{}).(RelationshipResultObserver); ok && observer != nil { - // RelatedNodes merges endpoints. Resolve each input independently to - // retain actual pair membership, then intersect with the final page. - page := map[string]bool{} - for _, entry := range entries { - page[entry.ID] = true - } relationships := []Relationship{} seen := map[Relationship]bool{} - for _, id := range ids { - related, lookupErr := k.RelatedNodes(ctx, keg.RelatedNodesOptions{NodeIDs: []keg.NodeId{id}, Direction: opts.Direction}) - if lookupErr != nil { - return nil, lookupErr + for _, pair := range related.Pairs { + source, parseErr := keg.ParseNode(pair.From) + if parseErr != nil || source == nil { + return nil, fmt.Errorf("invalid relationship source %q: %w", pair.From, keg.ErrInvalid) + } + other, parseErr := keg.ParseNodeRef(pair.To) + if parseErr != nil { + return nil, parseErr + } + target := other.Node + if other.Form == keg.RefQualified { + target.Alias = "@" + other.Namespace + "/" + other.KegName + } + rel := Relationship{Source: *source, Target: target} + if opts.Direction == keg.RelatedBacklinks { + rel.Source, rel.Target = rel.Target, rel.Source } - for _, entry := range related { - if !page[entry.ID] { - continue - } - other, parseErr := keg.ParseNodeRef(entry.ID) - if parseErr != nil { - return nil, parseErr - } - target := other.Node - if other.Form == keg.RefQualified { - target.Alias = "@" + other.Namespace + "/" + other.KegName - } - rel := Relationship{Source: id, Target: target} - if opts.Direction == keg.RelatedBacklinks { - rel.Source, rel.Target = rel.Target, rel.Source - } - pair := rel - if !seen[pair] { - seen[pair] = true - relationships = append(relationships, rel) - } + if !seen[rel] { + seen[rel] = true + relationships = append(relationships, rel) } } observer(ctx, k, relationships) @@ -467,34 +412,39 @@ func (t *Tap) Grep(ctx context.Context, opts GrepOptions) ([]string, error) { return []string{}, fmt.Errorf("unable to open keg: %w", err) } + formatted := opts.IdOnly || opts.Format != "" + var compiled compiledFormat + if formatted { + if compiled, err = compileListFormat(opts.Format); err != nil { + return []string{}, err + } + } + // The keg pages the matches and resolves the format's fields itself, so a + // formatted grep costs one call however many nodes match. kegMatches, err := k.Grep(ctx, keg.GrepOptions{ Pattern: opts.Query, IgnoreCase: opts.IgnoreCase, MaxLines: opts.MaxLines, + Fields: compiled.selectorTexts(opts.IdOnly), + Offset: opts.Offset, + Limit: opts.Limit, }) if err != nil { return []string{}, fmt.Errorf("invalid query regex %q: %w", opts.Query, err) } + if formatted { + rows := make([]keg.ListViewRow, len(kegMatches)) + for i, m := range kegMatches { + rows[i] = keg.ListViewRow{Entry: m.Entry, Fields: m.Fields} + } + return renderListView(compiled, rows, renderOptions{ + Format: opts.Format, IdOnly: opts.IdOnly, Reverse: opts.Reverse, + }), nil + } matches := make([]grepMatch, 0, len(kegMatches)) for _, m := range kegMatches { matches = append(matches, grepMatch{entry: m.Entry, lines: m.Lines}) } - - matches = applyOffsetSlice(matches, opts.Offset) - - if opts.Limit > 0 && len(matches) > opts.Limit { - matches = matches[:opts.Limit] - } - - matchedEntries := make([]keg.NodeIndexEntry, 0, len(matches)) - for _, match := range matches { - matchedEntries = append(matchedEntries, match.entry) - } - if opts.IdOnly || opts.Format != "" { - return t.renderNodeEntries(ctx, k, matchedEntries, renderOptions{ - Format: opts.Format, IdOnly: opts.IdOnly, Reverse: opts.Reverse, - }) - } return renderGrepMatches(matches, opts.Reverse), nil } @@ -507,53 +457,36 @@ func (t *Tap) Tags(ctx context.Context, opts TagsOptions) ([]string, error) { if err != nil { return []string{}, fmt.Errorf("unable to open keg: %w", err) } - listing, err := k.ListEntries(ctx, keg.ListEntriesOptions{Query: opts.Query}) + compiled, err := compileListFormat(opts.Format) + if err != nil { + return []string{}, err + } + queryExpr := strings.TrimSpace(opts.Query) + result, err := t.operations().Tags(ctx, k, tapapi.TagsRequest{ + Query: queryExpr, + Fields: compiled.selectorTexts(opts.IdOnly), + Offset: opts.Offset, + Limit: opts.Limit, + }) if err != nil { - if strings.TrimSpace(opts.Query) != "" { + if queryExpr != "" { return []string{}, fmt.Errorf("invalid query expression: %w", err) } return []string{}, fmt.Errorf("unable to list entries: %w", err) } - - queryExpr := strings.TrimSpace(opts.Query) - if queryExpr == "" { - tags := listing.Tags - sortStringsAsc(tags) - tags = applyOffsetStrings(tags, opts.Offset) - if opts.Limit > 0 && len(tags) > opts.Limit { - tags = tags[:opts.Limit] + tags := result.Tags + if tags == nil { + tags = []string{} } if opts.Reverse { reverseStrings(tags) } return tags, nil } - - matchedEntries := listing.Entries - if len(matchedEntries) == 0 { - return []string{}, nil - } - matchedIDs := make(map[string]struct{}, len(matchedEntries)*2) - for _, entry := range matchedEntries { - matchedIDs[entry.ID] = struct{}{} - if node, parseErr := keg.ParseNode(entry.ID); parseErr == nil && node != nil { - matchedIDs[node.Path()] = struct{}{} - } - } - - entries := matchedEntries - sortNodeIndexEntries(entries) - - entries = applyOffset(entries, opts.Offset) - - if opts.Limit > 0 && len(entries) > opts.Limit { - entries = entries[:opts.Limit] - } - - return t.renderNodeEntries(ctx, k, entries, renderOptions{ + return renderListView(compiled, result.Rows, renderOptions{ Format: opts.Format, IdOnly: opts.IdOnly, Reverse: opts.Reverse, - }) + }), nil } func grepContentLineMatches(re *regexp.Regexp, raw []byte) []string { @@ -612,94 +545,6 @@ type renderOptions struct { Reverse bool } -// enrichWarnThreshold is the number of nodes above which a format requiring -// per-node reads warns once, so a slow listing explains itself. -const enrichWarnThreshold = 200 - -// renderNodeEntries renders one line per entry using the compiled format. -// -// Formats naming only intrinsics, index timestamps, or legacy verbs — which -// includes the default — perform no additional I/O. A format naming metadata -// or a statistics field costs one read per node for each, so the whole pass is -// wrapped in a single keg read boundary: the boundary is exclusive and -// context-re-entrant, so without this the pass would acquire and release it -// twice per node and block every other process on the keg. -func (t *Tap) renderNodeEntries( - ctx context.Context, - k keg.Keg, - entries []keg.NodeIndexEntry, - opts renderOptions, -) ([]string, error) { - if opts.IdOnly { - return renderNodeIDs(entries, opts.Reverse), nil - } - - compiled, err := compileListFormat(opts.Format) - if err != nil { - return nil, err - } - - lines := make([]string, 0, len(entries)) - enrich := compiled.needsMeta || compiled.needsStats - - render := func(ctx context.Context) error { - start, end, step := iterationBounds(len(entries), opts.Reverse) - for i := start; i != end; i += step { - if enrich { - if err := ctx.Err(); err != nil { - return err - } - } - src := nodeFieldSource{entry: entries[i]} - if enrich { - t.loadNodeFields(ctx, k, &src, compiled) - } - lines = append(lines, expandFormat(compiled, src)) - } - return nil - } - - if !enrich { - if err := render(ctx); err != nil { - return nil, err - } - return lines, nil - } - - if len(entries) >= enrichWarnThreshold { - t.Runtime.Logger().Warn( - "listing format reads per-node metadata", - "nodes", len(entries), - "keg", describeKeg(k), - ) - } - if err := keg.WithReadBoundary(ctx, k, render); err != nil { - return nil, err - } - return lines, nil -} - -// loadNodeFields fetches only what the compiled format actually needs. A read -// failure leaves the value empty rather than failing the listing: the stale -// index path already anticipates nodes that are indexed but unreadable, and one -// broken node must not blank an entire listing. -func (t *Tap) loadNodeFields(ctx context.Context, k keg.Keg, src *nodeFieldSource, compiled compiledFormat) { - id, err := keg.ParseNode(src.entry.ID) - if err != nil || id == nil { - return - } - if compiled.needsMeta { - if meta, err := k.GetMeta(ctx, *id); err == nil { - src.meta = meta - } - } - if compiled.needsStats { - if stats, err := k.GetStats(ctx, *id); err == nil { - src.stats = stats - } - } -} - func renderNodeIDs(entries []keg.NodeIndexEntry, reverse bool) []string { lines := make([]string, 0, len(entries)) start, end, step := iterationBounds(len(entries), reverse) diff --git a/pkg/tapper/tap_list_format.go b/pkg/tapper/tap_list_format.go index 8c1313c3..1bbd9a29 100644 --- a/pkg/tapper/tap_list_format.go +++ b/pkg/tapper/tap_list_format.go @@ -196,14 +196,11 @@ func renderListView(compiled compiledFormat, rows []keg.ListViewRow, opts render // nodeFieldSource carries everything needed to expand a compiled format for // one node. // -// Values resolved by the server arrive already rendered in `resolved`, keyed by -// selector text. meta and stats are only populated on the fallback path, where -// the client had to read them itself. +// Values the keg resolved arrive already rendered in `resolved`, keyed by +// selector text; everything else comes from the index entry. type nodeFieldSource struct { entry keg.NodeIndexEntry resolved map[string]string - meta *keg.NodeMeta - stats *keg.NodeStats } // fieldValue resolves one selector against a node, preferring a value the @@ -216,7 +213,7 @@ func fieldValue(sel keg.FieldSelector, src nodeFieldSource) string { return value } } - return keg.FieldValue(sel, src.entry, src.meta, src.stats) + return keg.FieldValue(sel, src.entry, nil, nil) } // expandFormat renders one line for a node. diff --git a/pkg/tapper/tap_list_format_test.go b/pkg/tapper/tap_list_format_test.go index 7ccd18d1..ae708047 100644 --- a/pkg/tapper/tap_list_format_test.go +++ b/pkg/tapper/tap_list_format_test.go @@ -223,7 +223,7 @@ func TestExpandFormatAbsentMetaAndStatsRenderEmpty(t *testing.T) { func TestExpandFormatAccessCountRendersZero(t *testing.T) { // accessCount has no absent state on disk, so it always renders a number. now := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) - src := nodeFieldSource{entry: testEntry(), stats: keg.NewStats(now)} + src := resolvedSource(t, "%{.accessCount}", nil, keg.NewStats(now)) if got := renderOne(t, "%{.accessCount}", src); got != "0" { t.Errorf("accessCount = %q, want %q", got, "0") } @@ -236,7 +236,7 @@ func TestExpandFormatIntrinsicsShadowMetadata(t *testing.T) { if err := meta.Set(t.Context(), "title", "metadata title"); err != nil { t.Fatalf("set title: %v", err) } - src := nodeFieldSource{entry: testEntry(), meta: meta} + src := resolvedSource(t, "%{title}", meta, nil) if got := renderOne(t, "%{title}", src); got != "A Node" { t.Errorf("title = %q, want the intrinsic %q", got, "A Node") } @@ -286,3 +286,20 @@ func TestCompileListFormatRealTabIsUnaffected(t *testing.T) { t.Errorf("real tab = %q, want %q", got, "3\tA Node") } } + +// resolvedSource resolves a format's fields the way the keg does before +// rendering, so these tests exercise the renderer with keg-resolved values. +func resolvedSource(t *testing.T, format string, meta *keg.NodeMeta, stats *keg.NodeStats) nodeFieldSource { + t.Helper() + compiled, err := compileListFormat(format) + if err != nil { + t.Fatalf("compile %q: %v", format, err) + } + src := nodeFieldSource{entry: testEntry(), resolved: map[string]string{}} + for _, seg := range compiled.segments { + if seg.isField { + src.resolved[seg.sel.Text] = keg.FieldValue(seg.sel, src.entry, meta, stats) + } + } + return src +} diff --git a/pkg/tapper/tap_list_requests_test.go b/pkg/tapper/tap_list_requests_test.go new file mode 100644 index 00000000..9dea5f56 --- /dev/null +++ b/pkg/tapper/tap_list_requests_test.go @@ -0,0 +1,163 @@ +package tapper_test + +import ( + "context" + "encoding/json" + "net/http" + "strconv" + "sync" + "testing" + + "github.com/jlrickert/tapper/internal/testapi" + "github.com/jlrickert/tapper/pkg/keg" + "github.com/jlrickert/tapper/pkg/tapapi" + "github.com/jlrickert/tapper/pkg/tapper" + "github.com/stretchr/testify/require" +) + +// hubRecorder is a Hub stand-in that records every request path, so a test +// can assert what an operation costs in round trips. +type hubRecorder struct { + mu sync.Mutex + paths []string + bodies map[string]map[string]any +} + +func (h *hubRecorder) record(r *http.Request) { + h.mu.Lock() + defer h.mu.Unlock() + h.paths = append(h.paths, r.URL.Path) + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + if h.bodies == nil { + h.bodies = map[string]map[string]any{} + } + h.bodies[r.URL.Path] = body +} + +func (h *hubRecorder) recorded() []string { + h.mu.Lock() + defer h.mu.Unlock() + return append([]string(nil), h.paths...) +} + +func remoteTap(t *testing.T, handler http.HandlerFunc) (*tapper.Tap, *hubRecorder) { + t.Helper() + rec := &hubRecorder{} + srv := testapi.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rec.record(r) + w.Header().Set("Content-Type", "application/json") + handler(w, r) + })) + t.Cleanup(srv.Close) + sb := NewSandbox(t) + tap := newTestTap(t, sb) + k := keg.NewRemoteKeg(srv.URL, "token", sb.Runtime()) + tap.KegResolver = func(context.Context, tapper.KegTargetOptions, tapper.FlightRole) (keg.Keg, error) { return k, nil } + return tap, rec +} + +// A formatted grep over a Hub keg used to read metadata once per matching +// node, so one agent grep made hundreds of sequential requests (#123). The +// Hub pages the matches and resolves the format's fields itself. +func TestFormattedGrepIsOneRequest(t *testing.T) { + t.Parallel() + const matches = 200 + tap, rec := remoteTap(t, func(w http.ResponseWriter, r *http.Request) { + out := make([]keg.GrepMatch, matches) + for i := range out { + out[i] = keg.GrepMatch{ + Entry: keg.NodeIndexEntry{ID: strconv.Itoa(i), Title: "Node"}, + Lines: []string{"1:hit"}, + Fields: map[string]string{"type": "note"}, + } + } + _ = json.NewEncoder(w).Encode(map[string]any{"matches": out}) + }) + + lines, err := tap.Grep(t.Context(), tapper.GrepOptions{Query: "hit", Format: "%i %{type}", Limit: 500, Offset: 3}) + require.NoError(t, err) + require.Len(t, lines, matches) + require.Equal(t, "0 note", lines[0]) + + require.Equal(t, []string{"/grep"}, rec.recorded()) + body := rec.bodies["/grep"] + require.Equal(t, []any{"type"}, body["fields"]) + require.EqualValues(t, 500, body["limit"]) + require.EqualValues(t, 3, body["offset"]) +} + +// Links rendered with a metadata column resolve every row's fields in one +// listing call after the relationship lookup, never per node. +func TestFormattedLinksAreTwoRequests(t *testing.T) { + t.Parallel() + const related = 50 + tap, rec := remoteTap(t, func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/related": + out := keg.RelatedNodesResult{} + for i := 1; i <= related; i++ { + id := strconv.Itoa(i) + out.Entries = append(out.Entries, keg.NodeIndexEntry{ID: id, Title: "Node"}) + out.Pairs = append(out.Pairs, keg.RelatedPair{From: "0", To: id}) + } + _ = json.NewEncoder(w).Encode(out) + case "/list/view": + out := keg.ListViewResult{} + for i := 1; i <= related; i++ { + out.Rows = append(out.Rows, keg.ListViewRow{ + Entry: keg.NodeIndexEntry{ID: strconv.Itoa(i)}, + Fields: map[string]string{"type": "idea"}, + }) + } + _ = json.NewEncoder(w).Encode(out) + default: + w.WriteHeader(http.StatusNotFound) + } + }) + + lines, err := tap.Links(t.Context(), tapper.LinksOptions{NodeIDs: []string{"0"}, Format: "%i %{type}"}) + require.NoError(t, err) + require.Len(t, lines, related) + require.Equal(t, "1 idea", lines[0]) + require.Equal(t, []string{"/related", "/list/view"}, rec.recorded()) + require.Len(t, rec.bodies["/list/view"]["node_ids"], related) +} + +// A Hub keg answers links with its fields and relationship pairs in one +// request to the Hub's TAP operation. +func TestHubLinksAreOneTapRequest(t *testing.T) { + t.Parallel() + rec := &hubRecorder{} + srv := testapi.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rec.record(r) + w.Header().Set("Content-Type", "application/json") + if r.URL.Path != "/api/v1/tap/links" { + w.WriteHeader(http.StatusNotFound) + return + } + _ = json.NewEncoder(w).Encode(tapapi.RelatedResponse{ + Rows: []keg.ListViewRow{{Entry: keg.NodeIndexEntry{ID: "7", Title: "Seven"}, Fields: map[string]string{"type": "idea"}}}, + Pairs: []keg.RelatedPair{{From: "0", To: "7"}}, + Total: 1, + }) + })) + t.Cleanup(srv.Close) + sb := NewSandbox(t) + tap := newTestTap(t, sb) + k := keg.NewRemoteKeg(srv.URL+"/api/v1/@team/notes", "token", sb.Runtime()) + k.SetTarget(&keg.Target{Namespace: "team", KegName: "notes"}) + tap.KegResolver = func(context.Context, tapper.KegTargetOptions, tapper.FlightRole) (keg.Keg, error) { return k, nil } + + var observed []tapper.Relationship + ctx := tapper.WithRelationshipResultObserver(t.Context(), func(_ context.Context, _ keg.Keg, rows []tapper.Relationship) { observed = rows }) + lines, err := tap.Links(ctx, tapper.LinksOptions{NodeIDs: []string{"0"}, Format: "%i %{type}", Limit: 5}) + require.NoError(t, err) + require.Equal(t, []string{"7 idea"}, lines) + require.Equal(t, []string{"/api/v1/tap/links"}, rec.recorded()) + body := rec.bodies["/api/v1/tap/links"] + require.Equal(t, "@team/notes", body["keg"]) + require.Equal(t, []any{"type"}, body["fields"]) + require.EqualValues(t, 5, body["limit"]) + require.Equal(t, []tapper.Relationship{{Source: keg.NodeId{ID: 0}, Target: keg.NodeId{ID: 7}}}, observed) +}