syntax = "proto3";

package jammi.v1.pipeline;

import "jammi/v1/embedding.proto";
import "jammi/v1/inference.proto";
import "jammi/v1/trigger.proto";

// PipelineService — the wire surface for the engine's compute-to-data engine-
// state verbs that build durable graph/embedding artifacts or assemble a
// target's conditioning context.
//
// `BuildNeighborGraph`, `PropagateEmbeddings`, and `AssembleContext` are
// imperative session methods, not SQL queries, so they live as typed gRPC RPCs
// alongside SessionService / EmbeddingService / InferenceService / EvalService
// rather than on the Flight SQL surface (which carries query/result only, per
// ADR-01 §3.2). They are substrate primitives — materialising a k-NN graph,
// propagating features over a declared graph, or pooling a retrieval into a
// fixed-width context vector — and name no tenant-domain concept.
//
// Each verb delegates 1:1 to the engine session; the service reimplements no
// retrieval, graph, or aggregation logic — it is purely a wire adapter over the
// in-process engine. The two graph-build verbs transport only a result-table
// name (the compute stays server-side, so the materialised table is
// byte-identical to one built in-process); `AssembleContext` carries its pooled
// vector inline as IEEE-754 `float` (bit-exact for the engine's `Vec<f32>`).
//
// Tenant scope is read from the request's `SessionTenant` extension, set by
// the single async tenant-binding layer driven by the engine's
// `TenantResolver` (default: the `jammi-session-id` header → shared session
// store) — the one mechanism every gRPC service and Flight SQL bind through.
service PipelineService {
  // Materialise the k-nearest-neighbour graph of a source's embedding table and
  // return the new edge table. Peer of the engine's `build_neighbor_graph`; the
  // client reads the table via `sql(...)`.
  rpc BuildNeighborGraph(BuildNeighborGraphRequest) returns (jammi.v1.embedding.ResultTable);

  // Propagate an embedding table's features over a declared graph (the
  // decoupled-GNN forward pass) into a new, searchable embedding table. Peer of
  // the engine's `propagate_embeddings`; the client reads the table via
  // `sql(...)`.
  rpc PropagateEmbeddings(PropagateEmbeddingsRequest) returns (jammi.v1.embedding.ResultTable);

  // Assemble and encode a target's context set: retrieve `k` nearest neighbours
  // (or a declared-edge walk / their union), pair them with `value_columns`, and
  // pool the neighbour vectors permutation-invariantly into one fixed-width
  // context vector. Peer of the engine's `assemble_context`; returns the pooled
  // vector and the carried metadata inline (no durable handle).
  rpc AssembleContext(AssembleContextRequest) returns (AssembleContextResponse);

  // Assemble a point-in-time-correct table: for each row of `spine`, attach the
  // `facts` row valid as-of the spine row's temporal key, within each equality
  // group. Peer of the engine's `asof_join`; the compute stays server-side and
  // the client reads the materialised table via `sql(...)`.
  rpc AsofJoin(AsofJoinRequest) returns (jammi.v1.embedding.ResultTable);

  // Re-invoke a result table's recorded producer over the inputs' current state
  // (the action half of incremental recompute). Peer of the engine's
  // `recompute`: it reads the named table's recorded producing descriptor,
  // replays the producing verb, and — under `cascade = DOWNSTREAM` — sweeps every
  // transitive dependent once in topological order. The compute stays
  // server-side; the report names the recomputed tables (read via `sql(...)`)
  // and the transitive downstream-stale set.
  rpc Recompute(RecomputeRequest) returns (RecomputeReport);
}

// Whether a `Recompute` sweeps the bounded downstream DAG or only reports it.
// Mirrors the engine's `Cascade`; UNSPECIFIED maps to the engine default
// (`ReportOnly`) — the engine never re-runs a transitive dependent unless the
// caller explicitly asks.
enum Cascade {
  CASCADE_UNSPECIFIED = 0;
  // Recompute the named table only; report the transitive downstream-stale set
  // without recomputing any of it.
  CASCADE_REPORT_ONLY = 1;
  // One bounded topological sweep: recompute the named table, then every
  // transitive dependent in dependency order.
  CASCADE_DOWNSTREAM = 2;
}

// Re-invoke a result table's recorded producer. Mirrors the engine's
// `recompute(table, cascade)` binding: `table` is the result table to recompute;
// `cascade` selects report-only (default) or the one bounded downstream sweep.
message RecomputeRequest {
  // The result table to recompute, by registered name.
  string table = 1;
  // Whether to also sweep the bounded downstream DAG. UNSPECIFIED maps to the
  // engine default (`REPORT_ONLY`).
  Cascade cascade = 2;
}

// One table a recompute re-produced: the original name it replayed from, the new
// table the replay wrote, and the cache outcome of that replay (always
// `COMPUTED` — a recompute bypasses the cache). Mirrors the engine's
// `RecomputedTable`.
message RecomputedTable {
  // The original table name the recompute replayed.
  string original = 1;
  // The freshly materialised table the replay wrote.
  string recomputed = 2;
  // The cache outcome of the replay. Always `COMPUTED` for a recompute.
  jammi.v1.inference.CacheOutcome outcome = 3;
}

// The outcome of a recompute. Mirrors the engine's `RecomputeReport`: the tables
// re-produced (one for `REPORT_ONLY`; the named table plus its transitive
// dependents, in topological order, for `DOWNSTREAM`) and the transitive
// downstream-stale set (reported-not-acted-on for `REPORT_ONLY`; the set the
// sweep covered for `DOWNSTREAM`).
message RecomputeReport {
  // The tables re-produced, in recompute order.
  repeated RecomputedTable recomputed = 1;
  // The transitive set of tables downstream of the named table.
  repeated string downstream_stale = 2;
}

// Match direction for an as-of join. Mirrors the engine's `MatchDirection`;
// UNSPECIFIED maps to the engine default (`Backward`) — the only leakage-safe
// choice for past-keyed assembly.
enum AsofDirection {
  ASOF_DIRECTION_UNSPECIFIED = 0;
  ASOF_DIRECTION_BACKWARD = 1;
  ASOF_DIRECTION_FORWARD = 2;
  ASOF_DIRECTION_NEAREST = 3;
}

// Boundary inclusivity for an as-of join. Mirrors the engine's `Boundary`;
// UNSPECIFIED maps to the engine default (`Inclusive`, `<=`/`>=`).
enum AsofBoundary {
  ASOF_BOUNDARY_UNSPECIFIED = 0;
  ASOF_BOUNDARY_INCLUSIVE = 1;
  ASOF_BOUNDARY_EXCLUSIVE = 2;
}

// One side's column roles: the equality ("by") columns that partition the match
// into independent groups (may be empty — one global group), and the single
// temporal ordering column.
message AsofKey {
  // Equality columns, in declared order. May be empty.
  repeated string by = 1;
  // The temporal ordering column (required; exactly one).
  string time = 2;
}

// A look-back/forward limit for an as-of join: a candidate farther than this is
// no-match. Exactly one unit — `duration_micros` for temporal keys, `steps` for
// integer keys — modelled as a oneof so the unit is unambiguous; an absent
// `Tolerance` (the field unset on the request) means unbounded look-back.
message AsofTolerance {
  oneof limit {
    // Microsecond limit for a temporal key.
    int64 duration_micros = 1;
    // Step limit for an integer key.
    int64 steps = 2;
  }
}

// Run an as-of temporal join of two registered relations. Mirrors the engine's
// `asof_join` binding: `spine` and `facts` are source ids; the two `AsofKey`s
// name the equality + temporal columns on each side. `direction` and `boundary`
// default through their UNSPECIFIED arms; `tolerance` is optional (absent =
// unbounded); `tie_break_column` carries explicit presence — present names the
// secondary descending disambiguator, absent selects the loud error-on-ambiguity
// policy. `project` lists the right-side output columns (empty = all non-key).
message AsofJoinRequest {
  // The spine relation's source id.
  string spine = 1;
  // The facts relation's source id.
  string facts = 2;
  // The spine's column roles.
  AsofKey left = 3;
  // The facts' column roles.
  AsofKey right = 4;
  // Match direction. UNSPECIFIED maps to the engine default (`Backward`).
  AsofDirection direction = 5;
  // Boundary inclusivity. UNSPECIFIED maps to the engine default (`Inclusive`).
  AsofBoundary boundary = 6;
  // Optional look-back/forward limit; unset means unbounded.
  optional AsofTolerance tolerance = 7;
  // Optional tie-break secondary column (newest wins). Unset selects the loud
  // `AmbiguousMatch` policy — a true duplicate at the matched instant fails
  // rather than being resolved non-deterministically.
  optional string tie_break_column = 8;
  // Right-side columns to attach to the output. Empty keeps all non-key columns.
  repeated string project = 9;
}

// Materialise a source's k-NN similarity graph. Mirrors the engine's
// `build_neighbor_graph` binding: the count `k` is required; `min_similarity`,
// `mutual`, `exact`, and `table` are optional with explicit presence so an unset
// field resolves to the engine default rather than a literal `0`/`false`/`""`.
// The engine-side build knobs `self_exclude` / `exact_max_rows` / `resolve_keys`
// are not on the wire — the server applies the engine defaults, the single
// source of those values.
message BuildNeighborGraphRequest {
  // Source whose embedding table the graph is built over.
  string source_id = 1;
  // Neighbourhood size `k` (required).
  uint32 k = 2;
  // Optional minimum cosine similarity floor for an edge.
  optional float min_similarity = 3;
  // Keep only reciprocal edges.
  bool mutual = 4;
  // Use the exact (deterministic, complete) driver, gated by a row-count
  // ceiling; default false runs the approximate index-assisted driver.
  bool exact = 5;
  // Optional specific embedding table to build the graph over; unset resolves
  // the source's default embedding table.
  optional string table = 6;
  // Opt-in memoization. UNSPECIFIED maps to the engine default (`Bypass`). A
  // neighbour-graph is anchored on the immutable source-table digest, so it is
  // genuinely cacheable: a `Use` request reuses an exact prior build.
  jammi.v1.inference.CachePolicy cache = 7;
}

// Which neighbour-contribution normalisation a propagation hop uses. Mirrors the
// engine's `PropagationWeighting`; UNSPECIFIED maps to the engine default
// (`DegreeNormalized`).
enum PropagationWeighting {
  PROPAGATION_WEIGHTING_UNSPECIFIED = 0;
  PROPAGATION_WEIGHTING_DEGREE_NORMALIZED = 1;
  PROPAGATION_WEIGHTING_UNIFORM = 2;
  PROPAGATION_WEIGHTING_EDGE_SIMILARITY = 3;
}

// What a propagation emits. Mirrors the engine's `PropagationOutput`;
// UNSPECIFIED maps to the engine default (`Final`).
enum PropagationOutput {
  PROPAGATION_OUTPUT_UNSPECIFIED = 0;
  PROPAGATION_OUTPUT_FINAL = 1;
  PROPAGATION_OUTPUT_JUMPING_KNOWLEDGE = 2;
}

// A registered external edge source and its column bindings, for the
// `edge_source` arm of a propagation. The endpoints join to the source on the
// key; `weight_column` feeds the `edge_similarity` weighting.
message PropagateEdgeSource {
  // Registered source id holding the edge rows.
  string edge_source = 1;
  // Column holding the edge's source endpoint (default "src" when empty).
  string src_column = 2;
  // Column holding the edge's destination endpoint (default "dst" when empty).
  string dst_column = 3;
  // Optional column holding the edge weight.
  optional string weight_column = 4;
}

// Propagate an embedding table's features over a declared graph. Mirrors the
// engine's `propagate_embeddings` binding. The graph is either an S9 similarity
// graph (`edge_graph_table`, a `build_neighbor_graph` output) or a registered
// external edge source (`edge_source`) — exactly one, modelled as a `oneof` so
// both/neither is a decode error. The optional scalars carry explicit presence
// so an unset field resolves to the engine default.
message PropagateEmbeddingsRequest {
  // Source whose embedding table holds the input features `X⁽⁰⁾`.
  string source_id = 1;
  // Optional specific embedding table; unset resolves the source's default.
  optional string embedding_table = 2;
  // The graph to propagate over — exactly one arm.
  oneof graph {
    // An S9 similarity graph (a `build_neighbor_graph` output table name).
    string edge_graph_table = 3;
    // A registered external edge source + its column bindings.
    PropagateEdgeSource edge_source = 4;
  }
  // Edge direction read when building the neighbourhood. UNSPECIFIED maps to the
  // engine default (`Out`).
  jammi.v1.inference.EdgeDirection direction = 5;
  // Optional hop count (clamped to the engine hop cap).
  optional uint32 hops = 6;
  // Neighbour-contribution weighting.
  PropagationWeighting weighting = 7;
  // Optional APPNP teleport probability.
  optional double alpha = 8;
  // What the propagation emits.
  PropagationOutput output = 9;
  // Opt-in memoization. UNSPECIFIED maps to the engine default (`Bypass`). A
  // propagation is anchored on the immutable source-embedding-table and
  // edge-relation digests, so it is genuinely cacheable: a `Use` request reuses
  // an exact prior run.
  jammi.v1.inference.CachePolicy cache = 10;
}

// Which permutation-invariant reduction the context-set encoder pools with.
// Mirrors the engine's `SetAggregator`; UNSPECIFIED maps to the engine default
// (`Mean`).
enum SetAggregator {
  SET_AGGREGATOR_UNSPECIFIED = 0;
  SET_AGGREGATOR_MEAN = 1;
  SET_AGGREGATOR_SUM = 2;
  SET_AGGREGATOR_MAX = 3;
}

// Assemble and encode a target's context set. Mirrors the engine's
// `assemble_context` binding. `query` is the target's vector; an absent
// `edges` block is the ANN-only default, a present block (without `hybrid`) is a
// declared-edge walk, and a present block with `hybrid` is the union of both.
message AssembleContextRequest {
  // Source whose embedding table the context is retrieved against and whose rows
  // the members are pooled and hydrated from.
  string source_id = 1;
  // The target's query vector — the point whose neighbourhood is the context.
  // IEEE-754 `float` (bit-exact for the engine's `Vec<f32>`).
  repeated float query = 2;
  // Neighbourhood size `k`.
  uint32 k = 3;
  // Optional label/outcome columns hydrated for each context member.
  repeated string value_columns = 4;
  // The pooling reduction.
  SetAggregator aggregator = 5;
  // Drop the target's own row (`exclude_key`) before pooling — the leakage
  // guard. Explicit presence: an unset field resolves to the engine default
  // (`true`), the single source of that value, so a client that omits it keeps
  // the leakage guard on rather than silently disabling it via the proto3 bare-
  // bool wire default (`false`).
  optional bool exclude_self = 6;
  // Optional target row key dropped from the context when `exclude_self`.
  optional string exclude_key = 7;
  // Optional SQL predicate scoping the context to a split.
  optional string split = 8;
  // Optional declared-edge gather. Absent = ANN-only; present = declared-edge
  // walk, or (with `hybrid`) the union with the ANN context.
  optional jammi.v1.inference.EdgeGather edges = 9;
  // Union the declared-edge and ANN candidate sets (only meaningful with
  // `edges`).
  bool hybrid = 10;
}

// The pooled fixed-width context vector. A wrapper message (carried as an
// `optional` field) so a degenerate empty context (`None` — no neighbour
// survived exclusion/split) is distinguishable from a present-but-empty vector:
// the absence is a correctness signal a decoder reads as low-confidence, never
// silently averaged into a representation.
message ContextVector {
  // The pooled vector `ρ(Σ φ(xᵢ))`, in IEEE-754 `float`.
  repeated float values = 1;
}

// Result of one AssembleContext invocation. Mirrors the dict the engine binding
// returns: the pooled vector (presence-wrapped), the carried set size + member
// keys, the hydrated value rows, and the assembly-source fact.
message AssembleContextResponse {
  // The pooled context vector, or unset for a degenerate empty context.
  optional ContextVector context_vector = 1;
  // Number of neighbours that survived exclusion and entered the pool — carried
  // separately from the vector so a decoder uses the count without it corrupting
  // the pooled representation.
  uint64 context_size = 2;
  // Keys of the context members, in retrieval (descending-similarity) order.
  repeated string context_keys = 3;
  // The hydrated `value_columns` of the context members, as one Arrow IPC stream
  // (schema + batches) in `data_body`. Empty when no value columns were
  // requested. Reuses TriggerService's `ArrowBatch` shape.
  jammi.v1.trigger.ArrowBatch value_rows = 4;
  // How the context was assembled ("ann" / "edges" / "hybrid").
  string source = 5;
}
