diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6b0174cb..832ada17 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -248,6 +248,9 @@ jobs: run: make test-unit - name: Run architecture tests + # CI runs arch tests against the committed VENDORED rules facts + # (tests/arch/vendored/*). It never mounts the internal jentic-one-rules + # repo — the vendored-facts OSS-safety guard runs here regardless. run: make test-arch test-integration: diff --git a/.gitignore b/.gitignore index 0b882523..7de179bc 100644 --- a/.gitignore +++ b/.gitignore @@ -111,3 +111,9 @@ cli/AGENTS.md # Local toolchain (Go install for CLI builds) .toolchain/ + +# Private rules clone (jentic-one-rules) fetched by scripts/rules-clone.sh. Read +# at runtime (auto-detected here, or via JENTIC_RULES_DIR); its full rule prose + +# facts must NEVER be committed into this public repo. Contributors without access +# simply fall back to the vendored subset under tests/arch. +.rules/ diff --git a/.harness/setup.sh b/.harness/setup.sh index 72b9da47..75a24930 100755 --- a/.harness/setup.sh +++ b/.harness/setup.sh @@ -12,7 +12,7 @@ set -euo pipefail # HARNESS_ENV_FILE is a sourced shell file the harness uses to propagate # environment variables from this script back to subsequent harness steps. # We don't write to it yet, but we validate it up front so future setup logic -# can append `KEY=value` lines without surprise failures. +# can append KEY=value lines without surprise failures. : "${HARNESS_ENV_FILE:?HARNESS_ENV_FILE must be set by the harness}" if [[ ! -w "${HARNESS_ENV_FILE}" ]]; then echo "ERROR: HARNESS_ENV_FILE=${HARNESS_ENV_FILE} is not writable" >&2 @@ -25,4 +25,35 @@ if [[ "${1:-}" == "--teardown" ]]; then exit 0 fi +# Fetch the internal rules repo (read-only) so harness agents read the full rule +# guidance. The harness runlet lacks the ssh binary. Instead, the harness +# injects GITHUB_APP_TOKEN_FILE with a path to a refreshable token we can use. +# --teardown: harness owns compose lifecycle; nothing to do here. +echo "[harness] Fetching internal rules repo (jentic-one-rules)..." +RULES_DIR="$(pwd)/.rules" + +if [[ ! -d "${RULES_DIR}" ]]; then + # Grab the token from the harness's secure sidecar file + TOKEN="" + if [[ -n "${GITHUB_APP_TOKEN_FILE:-}" && -s "${GITHUB_APP_TOKEN_FILE}" ]]; then + TOKEN=$(cat "${GITHUB_APP_TOKEN_FILE}") + fi + + if [[ -n "${TOKEN}" ]]; then + # Use --quiet so we don't accidentally leak the token URL in trace logs + if git clone --quiet "https://x-access-token:${TOKEN}@github.com/jentic/jentic-one-rules.git" "${RULES_DIR}"; then + echo "[harness] rules-clone: Successfully cloned jentic-one-rules." + else + echo "[harness] rules-clone: failed to clone (expected for OSS users without access)." + fi + else + echo "[harness] rules-clone: No GITHUB_APP_TOKEN_FILE found. Skipping rules repo clone." + fi +fi + +# Export the mount so subsequent harness steps inherit it +if [[ -d "${RULES_DIR}" ]]; then + echo "JENTIC_RULES_DIR=${RULES_DIR}" >>"${HARNESS_ENV_FILE}" +fi + echo "ready" diff --git a/AGENTS.md b/AGENTS.md index 7528090d..ad165e9f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,5 +1,10 @@ # AGENTS.md +> **If `.rules/` exists, read `.rules/AGENTS.md` and use it instead of this file.** +> Otherwise use the guidance below. + +## In-repo guidance + This repo's agent guidance lives in **[`CLAUDE.md`](CLAUDE.md)**. Read it first. It covers the quick start, commands, project layout (backend + `ui/` frontend), @@ -19,3 +24,4 @@ Key conventions, in one breath: (`alwaysApply`). When `CLAUDE.md` and this file disagree, `CLAUDE.md` wins. + diff --git a/CLAUDE.md b/CLAUDE.md index fe440566..076b9a29 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,5 +1,8 @@ # CLAUDE.md +> **If `.rules/` exists, read `.rules/AGENTS.md` and use it instead of this file.** +> Otherwise use the guidance below. + ## Quick Start ```bash diff --git a/GEMINI.md b/GEMINI.md new file mode 100644 index 00000000..b17f7025 --- /dev/null +++ b/GEMINI.md @@ -0,0 +1,5 @@ +# GEMINI.md + +> **If `.rules/` exists, read `.rules/AGENTS.md` and use it instead of this file.** + +Otherwise, read [`AGENTS.md`](AGENTS.md) — this repo's canonical agent guidance. diff --git a/README.md b/README.md index f17b5877..db481644 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,21 @@ Tests are split into tiers: Commits follow [Conventional Commits](https://www.conventionalcommits.org/) with a mandatory scope, enforced repo-wide by a `commit-msg` hook. +### Full development rules (jentic maintainers) + +The architecture tests self-enforce against a small vendored subset of facts +([`tests/arch/vendored/`](tests/arch/vendored/)), so a plain clone needs nothing +extra. jentic maintainers with access to the internal `jentic-one-rules` repo can +pull the **full** rule guidance for local dev / agents: + +```bash +scripts/rules-clone.sh # read-only clone into .rules/ (gitignored, auto-detected) +``` + +Once present, the arch tests and coding agents pick up the live rules +automatically — no env var needed. If you lack access the script fails soft and +the vendored subset is used, so external contributors are unaffected. + ## Security & telemetry - **Credentials stay local.** Stored credentials are encrypted at rest and are only ever decrypted inside the Broker at execution time. They are never returned to callers, logged in cleartext, or exposed to the agent. diff --git a/cli/internal/cmd/app.go b/cli/internal/cmd/app.go index f821351e..f88ea47e 100644 --- a/cli/internal/cmd/app.go +++ b/cli/internal/cmd/app.go @@ -2,7 +2,6 @@ package cmd import ( "io" - "os" "github.com/jentic/jentic-one/cli/internal/config" ) @@ -10,6 +9,11 @@ import ( // App is the dependency container threaded into every command constructor. It // holds the resolved filesystem paths and the output streams, so commands carry // no package-global state and are constructible (and testable) in isolation. +// +// App is the internal wiring derived from the exported core.AppContainer +// (see root.go's appFromContainer): the container carries the injectable seams +// a downstream package can override, while App carries the resolved paths every +// subcommand needs. type App struct { // Paths resolves every filesystem location the CLI owns. Paths config.Paths @@ -17,12 +21,3 @@ type App struct { Out io.Writer Err io.Writer } - -// newApp builds the default application wiring (real paths, os streams). -func newApp() (*App, error) { - paths, err := config.NewPaths() - if err != nil { - return nil, err - } - return &App{Paths: paths, Out: os.Stdout, Err: os.Stderr}, nil -} diff --git a/cli/internal/cmd/root.go b/cli/internal/cmd/root.go index e3d50c67..5e78c713 100644 --- a/cli/internal/cmd/root.go +++ b/cli/internal/cmd/root.go @@ -10,6 +10,9 @@ import ( "syscall" "github.com/spf13/cobra" + + "github.com/jentic/jentic-one/cli/internal/config" + "github.com/jentic/jentic-one/cli/pkg/core" ) // Build-time version metadata. These are overridden via -ldflags @@ -141,14 +144,49 @@ func ExecuteAPI() { os.Exit(runRoot(newAPIRootCmd)) } -// runRoot builds the root command via build, wires a signal-cancelled context, -// and executes it. The real work lives here (rather than in Execute*) so that -// deferred cleanup (the signal-context cancel) always runs before the process -// exits. -func runRoot(build func(*App) *cobra.Command) int { - app, err := newApp() +// defaultContainer builds the default injection container (no extra commands). +// A downstream package builds its own core.AppContainer{ExtraCommands: ...} and +// calls core.NewRootCmd directly from its own main.go. +func defaultContainer() *core.AppContainer { + return &core.AppContainer{Out: os.Stdout, Err: os.Stderr} +} + +// appFromContainer derives the internal App (resolved paths + streams) from the +// injected container. Paths are resolved here — the exported core package stays +// free of the internal config package, keeping the dependency edge +// internal/cmd → pkg/core one-directional. +func appFromContainer(deps *core.AppContainer) (*App, error) { + paths, err := config.NewPaths() if err != nil { - fmt.Fprintln(os.Stderr, "error:", err) + return nil, err + } + return &App{Paths: paths, Out: deps.Out, Err: deps.Err}, nil +} + +// runRoot builds the root command via the built-in tree builder, wires a +// signal-cancelled context, and executes it. It composes the tree through +// core.NewRootCmd so any injected ExtraCommands are appended after the built-in +// set. The real work lives here (rather than in Execute*) so that deferred +// cleanup (the signal-context cancel) always runs before the process exits. +func runRoot(build func(*App) *cobra.Command) int { + deps := defaultContainer() + + // Adapt the internal (*App)-based tree builder to core.TreeBuilder. App + // construction (path resolution) can fail; surface it as a build-time panic + // captured below rather than threading an error through the cobra tree. + var buildErr error + tree := func(d *core.AppContainer) *cobra.Command { + app, err := appFromContainer(d) + if err != nil { + buildErr = err + return &cobra.Command{RunE: func(*cobra.Command, []string) error { return err }} + } + return build(app) + } + + root := core.NewRootCmd(deps, tree) + if buildErr != nil { + fmt.Fprintln(os.Stderr, "error:", buildErr) return 1 } @@ -157,7 +195,7 @@ func runRoot(build func(*App) *cobra.Command) int { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - err = build(app).ExecuteContext(ctx) + err := root.ExecuteContext(ctx) if err == nil { return 0 } diff --git a/cli/pkg/core/container.go b/cli/pkg/core/container.go new file mode 100644 index 00000000..eff02e4f --- /dev/null +++ b/cli/pkg/core/container.go @@ -0,0 +1,52 @@ +// Package core exposes the CLI's injectable dependency container and root +// builder so a downstream package can compose its own binaries without editing +// the built-in command tree. Everything here is importable (unlike internal/). +// +// Migration ordering is deliberately NOT modelled here: the Python runner +// (`python -m jentic_one.migrations.run`) owns the target set and its +// upgrade/rollback order via its DB_TARGETS registry. The CLI only invokes that +// runner (with a direction flag); it never maintains its own target list. +package core + +import ( + "io" + + "github.com/spf13/cobra" +) + +// CommandFactory builds a cobra command from the injected container. Extra +// command groups are supplied as factories so they are constructed against the +// same container (paths, streams) the built-in tree uses. +type CommandFactory func(deps *AppContainer) *cobra.Command + +// TreeBuilder builds the fully-configured root command for a binary from the +// injected container. internal/cmd supplies this so `core` never imports +// `internal/*` — which keeps the dependency edge one-directional +// (internal/cmd → pkg/core) and avoids an import cycle. +type TreeBuilder func(deps *AppContainer) *cobra.Command + +// AppContainer is the injected dependency set for the CLI command tree. The +// default binaries build a plain container; a downstream package builds its own +// and adds commands via ExtraCommands. +// +// It deliberately carries NO migration-target list (see the package doc). +type AppContainer struct { + // Out and Err are the standard output streams (overridable in tests). + Out io.Writer + Err io.Writer + + // ExtraCommands are extra command groups appended after the built-in tree. + // nil for the default binaries. + ExtraCommands []CommandFactory +} + +// NewRootCmd builds a root command tree using the injected container. `build` +// assembles the built-in command set (supplied by internal/cmd); any +// ExtraCommands are appended last so they never shadow built-in commands. +func NewRootCmd(deps *AppContainer, build TreeBuilder) *cobra.Command { + root := build(deps) + for _, f := range deps.ExtraCommands { + root.AddCommand(f(deps)) + } + return root +} diff --git a/cli/pkg/core/container_test.go b/cli/pkg/core/container_test.go new file mode 100644 index 00000000..4cf50ce7 --- /dev/null +++ b/cli/pkg/core/container_test.go @@ -0,0 +1,45 @@ +package core + +import ( + "io" + "testing" + + "github.com/spf13/cobra" +) + +// buildStub returns a minimal built-in-style tree builder for tests. +func buildStub(_ *AppContainer) *cobra.Command { + root := &cobra.Command{Use: "jentic"} + root.AddCommand(&cobra.Command{Use: "register"}) + return root +} + +func TestNewRootCmdBuildsTree(t *testing.T) { + deps := &AppContainer{Out: io.Discard, Err: io.Discard} + root := NewRootCmd(deps, buildStub) + + if root.Use != "jentic" { + t.Fatalf("root.Use = %q, want %q", root.Use, "jentic") + } + if _, _, err := root.Find([]string{"register"}); err != nil { + t.Fatalf("expected built-in 'register' command to be present: %v", err) + } +} + +func TestNewRootCmdAppendsExtraCommands(t *testing.T) { + deps := &AppContainer{ + Out: io.Discard, + Err: io.Discard, + ExtraCommands: []CommandFactory{ + func(_ *AppContainer) *cobra.Command { return &cobra.Command{Use: "proxy"} }, + func(_ *AppContainer) *cobra.Command { return &cobra.Command{Use: "trust"} }, + }, + } + root := NewRootCmd(deps, buildStub) + + for _, name := range []string{"register", "proxy", "trust"} { + if _, _, err := root.Find([]string{name}); err != nil { + t.Errorf("expected command %q to be present: %v", name, err) + } + } +} diff --git a/docker/local-setup/init-schemas.sql b/docker/local-setup/init-schemas.sql index db9aa7ea..4b855eb1 100644 --- a/docker/local-setup/init-schemas.sql +++ b/docker/local-setup/init-schemas.sql @@ -30,3 +30,23 @@ ALTER DEFAULT PRIVILEGES IN SCHEMA control GRANT ALL ON TYPES TO control_user; ALTER DEFAULT PRIVILEGES IN SCHEMA admin GRANT ALL ON TABLES TO admin_user; ALTER DEFAULT PRIVILEGES IN SCHEMA admin GRANT ALL ON SEQUENCES TO admin_user; ALTER DEFAULT PRIVILEGES IN SCHEMA admin GRANT ALL ON TYPES TO admin_user; + +-- ── Adding an extra schema that references these tables ───────────────────── +-- A downstream deployment can add its own isolated schema + role alongside the +-- ones above. If that schema's tables declare foreign keys into these tables, +-- the role needs USAGE on the referenced schema and REFERENCES on the target +-- tables — without them the FK creation fails with a permission error. The base +-- local-dev setup does not create any such schema (so the volume never needs +-- recreating for the default setup); wire the grants into that deployment's own +-- init script / DB-setup chart. Example shape: +-- +-- CREATE SCHEMA IF NOT EXISTS extra; +-- CREATE ROLE extra_user LOGIN PASSWORD 'extra_pass'; +-- GRANT USAGE, CREATE ON SCHEMA extra TO extra_user; +-- -- Grant on each referenced schema the extra tables FK into: +-- GRANT USAGE ON SCHEMA registry TO extra_user; +-- GRANT REFERENCES ON ALL TABLES IN SCHEMA registry TO extra_user; +-- ALTER DEFAULT PRIVILEGES IN SCHEMA registry +-- GRANT REFERENCES ON TABLES TO extra_user; +-- -- (repeat the three grants above for `control` / `admin` as needed) + diff --git a/docs/development/extending-jentic-one.md b/docs/development/extending-jentic-one.md new file mode 100644 index 00000000..b0697e63 --- /dev/null +++ b/docs/development/extending-jentic-one.md @@ -0,0 +1,154 @@ +# Extending jentic-one + +`jentic-one` ships a set of **backward-compatible seams** so an integrator can +inject alternate implementations and mount extra components **without editing +core code**. When no integrator wiring is supplied, behavior is identical to the +stock distribution. + +This guide is the unified composition story: how the seams fit together and the +order in which an integrator wires them. Each seam is also documented at its +definition — this page links them into one workflow. + +## The seams at a glance + +| Seam | Where | What it lets you do | +| ---- | ----- | ------------------- | +| `AppContainer` | `jentic_one.shared.web.container` | Inject a `Broker`; mount extra routers/installers after the built-in surfaces. | +| `register_config` | `jentic_one.shared.config` | Add a top-level config section validated by your own pydantic model. | +| `register_target` | `jentic_one.migrations.targets` | Add an isolated migration target to the ordered upgrade/rollback sequence. | +| `register_telemetry_event` | `jentic_one.shared.telemetry.events` | Forward extra telemetry events without editing the closed enum. | +| `jentic_one.testing` | `jentic_one.testing` | Compliance base classes that prove your implementations honor the seam contracts. | +| `pkg/core.AppContainer` | `cli/pkg/core` (Go) | Compose your own CLI binary with extra command groups. | + +All Python registries are process-global and populated **at import time** (e.g. +in your package's `__init__`) — call the `register_*` functions before +`load_config()` / app construction runs. + +## Composition workflow + +An integrator's composition root typically does the following, in order. + +### 1. Register config, migration targets, and telemetry events at import time + +```python +# my_ext/__init__.py +from pydantic import BaseModel + +from jentic_one.migrations.targets import MigrationTarget, register_target +from jentic_one.shared.config import register_config +from jentic_one.shared.db.base import RegistryBase # or your own declarative base +from jentic_one.shared.telemetry.events import register_telemetry_event + + +class MyExtConfig(BaseModel): + enabled: bool = False + endpoint: str = "https://example.internal" + + +# A new top-level YAML/env section: `my_ext:` in jentic-one.yaml. +# Collision guards reject names that shadow a core field (e.g. "broker") or the +# reserved "extensions" key. +register_config("my_ext", MyExtConfig) + +# An isolated migration target. Insertion order is the canonical UPGRADE order; +# rollback reverses it. Register after the built-ins so it upgrades last. +register_target(MigrationTarget("my_ext", RegistryBase.metadata)) + +# Forward an extra telemetry event. Wire names are lower_snake_case; collisions +# with the built-in enum/map are rejected so a typo can't silently drop events. +register_telemetry_event("my_ext.thing_happened", "thing_happened") +``` + +Read your validated config back with `AppConfig.extension("my_ext")`, which +returns your model instance (or `None` if the section is absent). + +### 2. Build an `AppContainer` and compose the app + +Start from the default container and add your `Broker` and any extra +routers/installers. Extra routers and installers run **after** all built-in +surfaces, so they mount last and never shadow a built-in route. + +```python +from fastapi import APIRouter + +from jentic_one.shared.context import Context +from jentic_one.shared.web.app_factory import create_combined_app +from jentic_one.shared.web.container import AppContainer + +from my_ext.broker import MyBroker + +my_router = APIRouter() # your extra routes + + +def build_app(ctx: Context): + container = AppContainer( + ctx=ctx, + broker=MyBroker(...), # injected data-plane broker + extra_routers=[(my_router, "/my-ext", ["my-ext"])], + extra_installers=[lambda app, ctx: ...], # runs against the root app last + ) + return create_combined_app(ctx, ctx.config.apps, container=container) +``` + +The container stashes your `broker` on `app.state.broker`. It is honored by +**both** callers of the "one pipeline, two callers" seam — the sync router +(`broker/web/routers/execute.py`) and the async worker +(`PipelineExecutor`) — so an injected broker reaches the sync **and** async +paths, not just one of them. + +> **Resilience tradeoff.** An injected `Broker` **owns its own transport and +> resilience** (circuit breaking, connection pooling, retries, per-host +> bulkheads, timeouts). It deliberately opts out of the built-in resilience +> stack that wraps the *default* broker's runner. If you only want to +> observe/enrich the standard path rather than replace transport, wrap a +> `DefaultBroker` and delegate to it so the built-in stack is retained. See the +> `Broker` protocol docstring in `jentic_one.shared.broker.broker`. + +### 3. Prove your implementations comply with the seam contracts + +`runtime_checkable` Protocols only validate method *presence* — an +implementation can drift in parameter names, defaults, or return type and still +pass `isinstance`. Subclass the `jentic_one.testing` compliance bases in your own +test suite to also assert the exact `inspect.signature` of every seam method: + +```python +# my_ext/tests/test_compliance.py +from jentic_one.shared.broker.broker import Broker +from jentic_one.testing import BaseBrokerComplianceTest, BaseSearchStrategyComplianceTest + +from my_ext.broker import MyBroker +from my_ext.search import MyStrategy + + +class TestMyBrokerCompliance(BaseBrokerComplianceTest): + def broker_factory(self) -> Broker: + return MyBroker(...) + + +class TestMyStrategyCompliance(BaseSearchStrategyComplianceTest): + strategy_cls = MyStrategy +``` + +These `Test*` subclasses are collected by pytest and fail loudly if your +implementation diverges from the built-in contract — the same guard the OSS +suite runs against its own defaults (`tests/unit/testing/test_compliance_oss.py`). + +### 4. (Optional) Compose your own CLI binary + +The Go CLI exposes an importable `pkg/core` container + root builder. Build your +own container with `ExtraCommands` and call `core.NewRootCmd`; the built-in +command tree is assembled by `internal/cmd` (the `internal/cmd → pkg/core` edge +stays one-directional, so an alternate binary composes its own tree without an +import cycle). Migration ordering is **not** modelled in Go — the CLI only +invokes the Python runner, which owns `DB_TARGETS` and its upgrade/rollback +order. + +## Breaking change: unknown config keys now fail loudly + +`AppConfig` sets `model_config = ConfigDict(extra="forbid")`. An **unrecognized +top-level config key** — one that is neither a core field nor a *registered* +extension section — now causes a **loud failure at startup** instead of being +silently ignored. This is defensively correct (it ensures extensions are +formally registered via `register_config`), but downstream configs with +legacy/typo top-level keys must be cleaned up or migrated to a registered +extension section before upgrading. diff --git a/scripts/rules-clone.sh b/scripts/rules-clone.sh new file mode 100755 index 00000000..ab855297 --- /dev/null +++ b/scripts/rules-clone.sh @@ -0,0 +1,66 @@ +#!/usr/bin/env bash +# +# rules-clone.sh — fetch the internal jentic-one-rules repo for local use. +# +# The public jentic-one repo self-enforces against a small VENDORED subset of +# machine facts (tests/arch/vendored/orm.facts.yaml). Maintainers with access to +# jentic-one-rules can additionally read the FULL rule prose + live facts at +# runtime by cloning that repo here (auto-detected — no env var needed). +# +# This clone is READ-ONLY in intent: it is placed in a gitignored path (.rules/) +# so private rule content can NEVER be committed into this public repo, and the +# facts loader (tests/arch/_rules_facts.py) only ever reads the files. +# +# Contributors without access are unaffected: the script fails SOFT (warns and +# exits 0) so setup flows never break for people who lack access. + +set -uo pipefail + +REPO_ROOT="$(cd "$(dirname "$0")/.." && pwd)" + +# Overridable so forks / mirrors don't hardcode the name in one load-bearing spot. +RULES_REMOTE="${JENTIC_RULES_REMOTE:-git@github.com:jentic/jentic-one-rules.git}" +RULES_REF="${JENTIC_RULES_REF:-main}" +MOUNT_DIR="${JENTIC_RULES_DIR:-${REPO_ROOT}/.rules}" + +warn() { printf 'rules-clone: %s\n' "$*" >&2; } + +if ! command -v git >/dev/null 2>&1; then + warn "git not found; skipping rules clone (OSS clones use the vendored facts)." + exit 0 +fi + +if [ -d "${MOUNT_DIR}/.git" ]; then + warn "updating existing rules clone at ${MOUNT_DIR}" + if ! git -C "${MOUNT_DIR}" fetch --quiet origin "${RULES_REF}" \ + || ! git -C "${MOUNT_DIR}" checkout --quiet "${RULES_REF}" \ + || ! git -C "${MOUNT_DIR}" reset --hard --quiet "origin/${RULES_REF}"; then + warn "could not update ${MOUNT_DIR}; leaving existing checkout in place." + fi +else + warn "cloning ${RULES_REMOTE} (ref ${RULES_REF}) into ${MOUNT_DIR}" + if ! git clone --quiet --depth 1 --branch "${RULES_REF}" "${RULES_REMOTE}" "${MOUNT_DIR}"; then + warn "no access to ${RULES_REMOTE} (this is expected for OSS users)." + warn "the repo self-enforces against tests/arch/vendored/*; nothing to do." + exit 0 + fi +fi + +if [ "${MOUNT_DIR}" = "${REPO_ROOT}/.rules" ]; then + cat < str: actor_type=ActorType.USER, actor_id=user.id, reason="operator password reset", + origin=None, ) logger.info("password_reset", user_id=user.id, email=email) diff --git a/src/jentic_one/broker/adapters/runners/base.py b/src/jentic_one/broker/adapters/runners/base.py index 70060eab..efe10cb0 100644 --- a/src/jentic_one/broker/adapters/runners/base.py +++ b/src/jentic_one/broker/adapters/runners/base.py @@ -13,52 +13,28 @@ from __future__ import annotations -from collections.abc import AsyncIterator -from contextlib import AbstractAsyncContextManager -from dataclasses import dataclass, field from typing import Protocol, runtime_checkable +from jentic_one.shared.broker.execution import ( + RunnerRequest, + RunnerResult, + StreamingResult, + StreamingUpstreamRunner, + UpstreamRunner, +) from jentic_one.shared.broker.protocols import RunnerCapabilities, Verb from jentic_one.shared.models.credentials import CredentialType - -@dataclass(frozen=True, slots=True) -class RunnerRequest: - """A transport-agnostic upstream request handed to a runner.""" - - method: str - url: str - headers: dict[str, str] = field(default_factory=dict) - body: bytes | None = None - timeout_s: float = 30.0 - - -@dataclass(frozen=True, slots=True) -class RunnerResult: - """The runner's response — the real upstream status/headers/body, verbatim.""" - - status_code: int - body: bytes - headers: dict[str, str] - content_type: str | None - duration_ms: int - - -@dataclass(frozen=True, slots=True) -class StreamingResult: - """A streamed upstream response: status/headers known up front, body lazy. - - Unlike :class:`RunnerResult` the body is **not** materialised — ``aiter`` is - the raw (still-compressed) upstream byte stream, consumed by the web edge to - feed a ``StreamingResponse``. It is only valid for the lifetime of the - ``stream()`` context manager that produced it; the upstream connection is - torn down when that context exits. - """ - - status_code: int - headers: dict[str, str] - content_type: str | None - aiter: AsyncIterator[bytes] +__all__ = [ + "HTTP_RUNNER_CAPABILITIES", + "CapabilityAwareRunner", + "RunnerRequest", + "RunnerResult", + "StreamingResult", + "StreamingUpstreamRunner", + "UpstreamRunner", + "capabilities_of", +] # The capability profile of the default HTTP runner. HTTP is the broker's core @@ -78,18 +54,6 @@ class StreamingResult: ) -@runtime_checkable -class UpstreamRunner(Protocol): - """Executes a single upstream request and returns its verbatim result. - - Decorators (retry/circuit/deadline/idempotency) implement this same protocol - and wrap a base runner, so the pipeline composes them without the handler - knowing which capabilities are active. - """ - - async def run(self, request: RunnerRequest) -> RunnerResult: ... - - @runtime_checkable class CapabilityAwareRunner(UpstreamRunner, Protocol): """An ``UpstreamRunner`` that declares its :class:`RunnerCapabilities`. @@ -123,19 +87,3 @@ def capabilities_of(runner: UpstreamRunner) -> RunnerCapabilities: supports_idempotency=False, supports_retries=False, ) - - -@runtime_checkable -class StreamingUpstreamRunner(UpstreamRunner, Protocol): - """An ``UpstreamRunner`` that can also stream the body without buffering (§08 E2.4). - - ``stream`` is an **async context manager**: it dispatches the request, yields - a :class:`StreamingResult` once the status/headers are in, and — critically — - holds the upstream ``httpx`` response open only for the duration of the - ``async with``. When the context exits (normal completion, size-cap/deadline - abort, or a client-disconnect ``CancelledError`` propagating out of the - body generator) the upstream stream is ``aclose()``d, releasing the pool slot - rather than leaking a zombie background drain. - """ - - def stream(self, request: RunnerRequest) -> AbstractAsyncContextManager[StreamingResult]: ... diff --git a/src/jentic_one/broker/core/exceptions.py b/src/jentic_one/broker/core/exceptions.py index 44bfc414..9abda84e 100644 --- a/src/jentic_one/broker/core/exceptions.py +++ b/src/jentic_one/broker/core/exceptions.py @@ -14,21 +14,11 @@ from __future__ import annotations -from enum import StrEnum from typing import Any, Literal from pydantic import BaseModel - -class ErrorOrigin(StrEnum): - """Disambiguates *who* failed on an otherwise-identical 5xx/503.""" - - BROKER = "broker" - """The broker/system failed (overloaded, misconfigured, bad request shape).""" - - UPSTREAM = "upstream" - """The external vendor failed (down, rate-limited, returned an error).""" - +from jentic_one.shared.broker.execution import ErrorOrigin as ErrorOrigin AgentStrategy = Literal[ "wait", diff --git a/src/jentic_one/broker/core/schemas.py b/src/jentic_one/broker/core/schemas.py index 49ebe393..da1e783f 100644 --- a/src/jentic_one/broker/core/schemas.py +++ b/src/jentic_one/broker/core/schemas.py @@ -1,30 +1,23 @@ -"""Broker execute request/response models.""" +"""Broker execute request/response models. -from __future__ import annotations +``ExecuteRequestContext`` is defined in ``shared/broker/schemas.py`` (it is part of +the :class:`~jentic_one.shared.broker.broker.Broker` seam contract) and re-exported +here so existing broker-internal imports keep working unchanged. The +``AsyncQueuedResponse*`` models are web-response bodies specific to the broker +surface and stay here. +""" -from typing import Any +from __future__ import annotations from pydantic import BaseModel, Field +from jentic_one.shared.broker.schemas import ExecuteRequestContext -class ExecuteRequestContext(BaseModel): - """Contextual metadata for a broker proxy request — discovery-driven. - - ``toolkit_id`` is no longer a required inbound header: it is derived (§03) or - supplied as an inbound disambiguator. ``operation_id`` / ``api_*`` come from - in-process discovery, not inbound ``Jentic-Api-*`` headers. - """ - - upstream_url: str - method: str - trace_id: str - toolkit_id: str | None = None - operation_id: str | None = None - api_vendor: str | None = None - api_name: str | None = None - api_version: str | None = None - prefer: str | None = None - pinned_revisions: dict[str, Any] | None = None +__all__ = [ + "AsyncQueuedResponse", + "AsyncQueuedResponseLinks", + "ExecuteRequestContext", +] class AsyncQueuedResponseLinks(BaseModel): diff --git a/src/jentic_one/broker/default_broker.py b/src/jentic_one/broker/default_broker.py new file mode 100644 index 00000000..f9b15312 --- /dev/null +++ b/src/jentic_one/broker/default_broker.py @@ -0,0 +1,53 @@ +"""Default Broker: adapts BrokerExecutionPipeline to the Broker seam. + +Holds no new behavior — it exists so the broker surface depends on the +:class:`~jentic_one.shared.broker.broker.Broker` protocol (not the concrete +pipeline), and so a downstream package can wrap or replace it (e.g. a wrapper +that delegates to a ``DefaultBroker`` for the standard path). +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable + +from fastapi import Response + +from jentic_one.broker.services.execution.pipeline import BrokerExecutionPipeline +from jentic_one.broker.web.streaming import open_streaming_response +from jentic_one.shared.broker.execution import ( + ExecutionContext, + ExecutionOutcome, + RunnerRequest, + StreamingOutcome, + StreamingUpstreamRunner, +) +from jentic_one.shared.broker.schemas import ExecuteRequestContext + + +class DefaultBroker: + """Thin adapter wrapping the existing execution pipeline (implements ``Broker``).""" + + def __init__(self, pipeline: BrokerExecutionPipeline) -> None: + self._pipeline = pipeline + + async def execute(self, request: RunnerRequest, context: ExecutionContext) -> ExecutionOutcome: + return await self._pipeline.execute(request, context) + + async def execute_streaming( + self, + runner: StreamingUpstreamRunner, + request: RunnerRequest, + ctx_req: ExecuteRequestContext, + execution_id: str, + *, + transfer_deadline_s: float, + background_callback: Callable[[StreamingOutcome], Awaitable[None]] | None = None, + ) -> Response: + return await open_streaming_response( + runner, + request, + ctx_req, + execution_id, + transfer_deadline_s=transfer_deadline_s, + background_callback=background_callback, + ) diff --git a/src/jentic_one/broker/services/execution/executor.py b/src/jentic_one/broker/services/execution/executor.py index c5af6f83..17058531 100644 --- a/src/jentic_one/broker/services/execution/executor.py +++ b/src/jentic_one/broker/services/execution/executor.py @@ -18,11 +18,14 @@ from __future__ import annotations +from collections.abc import Callable from typing import Any +from jentic_one.broker.adapters.runners.base import UpstreamRunner from jentic_one.broker.adapters.runners.registry import RunnerRegistry from jentic_one.broker.core.schemas import ExecuteRequestContext -from jentic_one.broker.services.execution.service import default_pipeline, run_execution +from jentic_one.broker.services.execution.service import default_broker, run_execution +from jentic_one.shared.broker.broker import Broker from jentic_one.shared.jobs.protocols import ( UpstreamExecRequest, UpstreamExecResult, @@ -31,19 +34,27 @@ class PipelineExecutor(UpstreamExecutor): - """Adapts the shared ``BrokerExecutionPipeline`` to the worker's protocol. + """Adapts the shared ``Broker`` to the worker's protocol. Selects the runner for each call through the shared :class:`RunnerRegistry` (the same seam the sync router uses), so the circuit-breaker latch, per-host bulkhead, and connection pool are shared across sync + async — and a non-HTTP scheme routes to its runner once one is registered. Each call runs through - ``run_execution`` — which dispatches the runner, folds the post-response - stages, and persists the ``executions`` row. The worker keeps only its - job-result + lifecycle-event writes. + ``run_execution`` against a :class:`Broker` built per request by + ``broker_factory`` (default: :func:`default_broker`; a caller may inject its + own), + which dispatches the runner, folds the post-response stages, and persists the + ``executions`` row. The worker keeps only its job-result + lifecycle-event writes. """ - def __init__(self, registry: RunnerRegistry) -> None: + def __init__( + self, + registry: RunnerRegistry, + *, + broker_factory: Callable[[UpstreamRunner], Broker] = default_broker, + ) -> None: self._registry = registry + self._broker_factory = broker_factory async def execute(self, request: UpstreamExecRequest, *, session: Any) -> UpstreamExecResult: ctx_req = _ctx_from_metadata(request) @@ -54,7 +65,7 @@ async def execute(self, request: UpstreamExecRequest, *, session: Any) -> Upstre headers=request.headers, session=session, timeout=request.timeout_s, - pipeline=default_pipeline(runner), + broker=self._broker_factory(runner), execution_id=request.metadata.get("execution_id"), actor_id=request.metadata["actor_id"], actor_type=request.metadata["actor_type"], diff --git a/src/jentic_one/broker/services/execution/pipeline.py b/src/jentic_one/broker/services/execution/pipeline.py index f3c0e0b3..73679240 100644 --- a/src/jentic_one/broker/services/execution/pipeline.py +++ b/src/jentic_one/broker/services/execution/pipeline.py @@ -24,41 +24,28 @@ from __future__ import annotations from collections.abc import Sequence -from dataclasses import dataclass, replace +from dataclasses import replace from typing import Protocol, runtime_checkable from jentic_one.broker.adapters.runners.base import ( RunnerRequest, - RunnerResult, UpstreamRunner, capabilities_of, ) from jentic_one.broker.adapters.runners.deadline import DeadlineRunner from jentic_one.broker.adapters.runners.retry import RetryRunner -from jentic_one.broker.core.exceptions import ErrorOrigin +from jentic_one.shared.broker.execution import ErrorOrigin, ExecutionContext, ExecutionOutcome from jentic_one.shared.broker.protocols import RunnerCapabilities from jentic_one.shared.config import RetryConfig -from jentic_one.shared.schemas import APIReference - -@dataclass(frozen=True, slots=True) -class ExecutionContext: - """Identity + discovery metadata threaded through the pipeline for one call.""" - - execution_id: str - toolkit_id: str | None - operation_id: str | None - api: APIReference | None - trace_id: str - - -@dataclass(frozen=True, slots=True) -class ExecutionOutcome: - """The immutable result a post-response stage may enrich (never the 2xx body).""" - - result: RunnerResult - context: ExecutionContext - error_origin: ErrorOrigin | None = None +__all__ = [ + "BrokerExecutionPipeline", + "ExecutionContext", + "ExecutionOutcome", + "PostResponseStage", + "build_runner", + "enrich_error_origin", +] @runtime_checkable diff --git a/src/jentic_one/broker/services/execution/service.py b/src/jentic_one/broker/services/execution/service.py index 14928680..f9e338b9 100644 --- a/src/jentic_one/broker/services/execution/service.py +++ b/src/jentic_one/broker/services/execution/service.py @@ -22,11 +22,13 @@ from jentic_one.broker.core.exceptions import BrokerError, CircuitOpenError from jentic_one.broker.core.execution import mint_execution_id from jentic_one.broker.core.schemas import ExecuteRequestContext +from jentic_one.broker.default_broker import DefaultBroker from jentic_one.broker.services.execution.pipeline import ( BrokerExecutionPipeline, ExecutionContext, ExecutionOutcome, ) +from jentic_one.shared.broker.broker import Broker from jentic_one.shared.config import SecurityConfig from jentic_one.shared.events import emit_event from jentic_one.shared.events.repeated_failure import maybe_emit_repeated_failure @@ -91,6 +93,17 @@ def default_pipeline(runner: UpstreamRunner) -> BrokerExecutionPipeline: return BrokerExecutionPipeline(runner) +def default_broker(runner: UpstreamRunner) -> Broker: + """Build the default :class:`Broker` for a runner. + + The per-request factory the surface + worker use when no ``Broker`` is + injected via the ``AppContainer``. Wraps :func:`default_pipeline` in a + :class:`DefaultBroker` so the execution path depends on the neutral ``Broker`` + seam; a caller swaps this factory to inject its own implementation. + """ + return DefaultBroker(default_pipeline(runner)) + + def _api_reference(ctx_req: ExecuteRequestContext) -> APIReference | None: if not ctx_req.api_vendor: return None @@ -108,19 +121,20 @@ async def run_execution( headers: dict[str, str] | None, session: Any, timeout: float = 30.0, - pipeline: BrokerExecutionPipeline, + broker: Broker, execution_id: str | None = None, actor_id: str, actor_type: str, origin: str | None = None, security_config: SecurityConfig | None = None, ) -> ExecutionOutcome: - """Run the upstream call through the shared pipeline and persist the record. + """Run the upstream call through the injected ``Broker`` and persist the record. - On a transport-level failure the pipeline's runner raises a ``BrokerError``; + On a transport-level failure the broker's pipeline raises a ``BrokerError``; we persist a FAILED record before re-raising so the central handler can map - it to problem+json. The ``pipeline`` (and thus the shared upstream client it - wraps) is supplied by the caller (§04 — one client per process). + it to problem+json. The ``broker`` (and thus the shared upstream client it + wraps) is supplied by the caller (§04 — one client per process); the default + builds a :class:`DefaultBroker` per request, a caller may inject its own. ``execution_id`` lets the async worker reuse the id already handed to the client in the ``202`` (and used as the job's correlation id) so the persisted @@ -168,7 +182,7 @@ async def run_execution( span.set_attribute("toolkit_id", ctx_req.toolkit_id or "") span.set_attribute("api_vendor", ctx_req.api_vendor or "") with jentic_tracestate(tracestate_member): - outcome = await pipeline.execute(runner_request, exec_context) + outcome = await broker.execute(runner_request, exec_context) except BrokerError as exc: logger.error("execution_failed", execution_id=execution_id, error=exc.detail[:128]) await _persist( @@ -287,20 +301,20 @@ async def execute_upstream( headers: dict[str, str] | None = None, session: Any, timeout: float = 30.0, - pipeline: BrokerExecutionPipeline, + broker: Broker, actor_id: str, actor_type: str, origin: str | None = None, security_config: SecurityConfig | None = None, ) -> RunnerResult: - """Run the pipeline and return only the upstream result (status/headers/body).""" + """Run the broker and return only the upstream result (status/headers/body).""" outcome = await run_execution( ctx_req, body=body, headers=headers, session=session, timeout=timeout, - pipeline=pipeline, + broker=broker, actor_id=actor_id, actor_type=actor_type, origin=origin, diff --git a/src/jentic_one/broker/web/app.py b/src/jentic_one/broker/web/app.py index 4ca07e54..76bf3dba 100644 --- a/src/jentic_one/broker/web/app.py +++ b/src/jentic_one/broker/web/app.py @@ -33,6 +33,7 @@ from jentic_one.shared.state import build_state_backend from jentic_one.shared.tracing import instrument_outbound_client from jentic_one.shared.web.app_factory import create_surface_app +from jentic_one.shared.web.container import AppContainer from jentic_one.shared.web.health import make_health_router @@ -51,8 +52,17 @@ def _routers(*, readiness_saturation_threshold: float) -> list[tuple[APIRouter, ] -def create_app(ctx: Context) -> FastAPI: - """Create the broker FastAPI application for standalone deployment.""" +def create_app(ctx: Context, container: AppContainer | None = None) -> FastAPI: + """Create the broker FastAPI application for standalone deployment. + + ``container`` is the DI seam: when omitted the default is used and behavior is + unchanged. A caller passes its own container to inject a ``Broker`` — it is + threaded to :func:`create_surface_app`, which sets ``app.state.broker`` and + ``app.state.broker_factory`` so the injected broker reaches BOTH the sync + router and the async worker (the worker's ``PipelineExecutor`` reads + ``broker_factory`` in ``broker_lifespan`` below). + """ + container = container or AppContainer.default(ctx) resilience = ctx.config.broker.resilience upstream_cfg = resilience.upstream meter = get_meter("broker") @@ -155,7 +165,18 @@ def _observe_in_flight(_options: CallbackOptions) -> list[Observation]: # resolves credentials with the same CredentialService. Both are stashed # on app.state so the shared/ worker factory wires them without importing # broker/ (the arch boundary). Built here so they wrap the live runner. - app.state.broker_upstream_executor = PipelineExecutor(registry) + # + # Seam symmetry ("one pipeline, two callers"): if a caller set a + # `broker_factory` on app.state (the same factory the sync router honors + # in web/routers/execute.py), the worker's executor uses it too — so an + # injected Broker reaches BOTH the sync and async paths, not just sync. + # Unset by default → PipelineExecutor falls back to `default_broker`. + injected_broker_factory = getattr(app.state, "broker_factory", None) + app.state.broker_upstream_executor = ( + PipelineExecutor(registry, broker_factory=injected_broker_factory) + if injected_broker_factory is not None + else PipelineExecutor(registry) + ) app.state.broker_credential_injector = CredentialService(ctx) try: yield @@ -175,6 +196,7 @@ def _observe_in_flight(_options: CallbackOptions) -> list[Observation]: title="jentic-one-broker", routers=_routers(readiness_saturation_threshold=resilience.readiness_saturation_threshold), extra_lifespan=broker_lifespan, + container=container, ) # One admission gate shared by the shedding middleware and the readiness # probe (§05 R5.2), so both observe the same in-flight counter. diff --git a/src/jentic_one/broker/web/routers/execute.py b/src/jentic_one/broker/web/routers/execute.py index cad9bb47..25c5140c 100644 --- a/src/jentic_one/broker/web/routers/execute.py +++ b/src/jentic_one/broker/web/routers/execute.py @@ -62,7 +62,7 @@ from jentic_one.broker.services.discovery import discover, resolve_pin_for_api from jentic_one.broker.services.execution.pipeline import ExecutionOutcome from jentic_one.broker.services.execution.service import ( - default_pipeline, + default_broker, persist_streaming_execution, run_execution, ) @@ -78,8 +78,9 @@ RuleEvaluatorDep, ToolkitDeriver, ) -from jentic_one.broker.web.streaming import StreamingOutcome, open_streaming_response +from jentic_one.broker.web.streaming import StreamingOutcome from jentic_one.shared.auth.identity import Identity +from jentic_one.shared.broker.broker import Broker from jentic_one.shared.broker.protocols import ( RegistryResolverProtocol, ResolveResult, @@ -411,6 +412,20 @@ def _apply_injection( return upstream_url, headers +def _resolve_broker(request: Request, runner: UpstreamRunner) -> Broker: + """Select the broker for this request: an injected instance wins over the default. + + An injected ``app.state.broker`` owns its own transport and is used verbatim; + only its absence falls back to the per-request ``broker_factory`` (default: + :func:`default_broker`) over the selected runner. Both the buffered and + streaming sync paths resolve through here so neither can bypass an injected + broker's controls. + """ + injected = getattr(request.app.state, "broker", None) + broker_factory = getattr(request.app.state, "broker_factory", default_broker) + return injected if injected is not None else broker_factory(runner) + + async def _handle( request: Request, method: str, @@ -567,13 +582,14 @@ async def _handle( forwarded = forward_headers(request.headers, auth_headers) async with ctx.admin_db.transaction() as session: + broker = _resolve_broker(request, runner) outcome = await run_execution( ctx_req, body=body, headers=forwarded, session=session, timeout=ctx.config.broker.upstream_timeout_s, - pipeline=default_pipeline(runner), + broker=broker, actor_id=identity.sub, actor_type=identity.actor_type.value, origin=identity.origin.value, @@ -644,7 +660,8 @@ async def _persist_callback(outcome: StreamingOutcome) -> None: origin=identity.origin.value, ) - return await open_streaming_response( + broker = _resolve_broker(request, runner) + return await broker.execute_streaming( runner, runner_request, ctx_req, diff --git a/src/jentic_one/broker/web/streaming.py b/src/jentic_one/broker/web/streaming.py index 565c62a9..cbd7661a 100644 --- a/src/jentic_one/broker/web/streaming.py +++ b/src/jentic_one/broker/web/streaming.py @@ -27,7 +27,6 @@ import time from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import AsyncExitStack -from dataclasses import dataclass, field from fastapi import Response from starlette.background import BackgroundTask @@ -42,18 +41,13 @@ from jentic_one.broker.core.headers import JenticHeader from jentic_one.broker.core.proxy_headers import passthrough_streaming_headers from jentic_one.broker.core.schemas import ExecuteRequestContext +from jentic_one.shared.broker.execution import StreamingOutcome - -@dataclass -class StreamingOutcome: - """Captures the final state of a streaming execution for persistence.""" - - execution_id: str - http_status: int - started_at_perf: float = field(default_factory=time.perf_counter) - duration_ms: int = 0 - error: str | None = None - bytes_transferred: int = 0 +__all__ = [ + "StreamingOutcome", + "guarded_body", + "open_streaming_response", +] async def guarded_body( diff --git a/src/jentic_one/migrations/env.py b/src/jentic_one/migrations/env.py index 12ede9c1..8067bdbd 100644 --- a/src/jentic_one/migrations/env.py +++ b/src/jentic_one/migrations/env.py @@ -16,7 +16,7 @@ from sqlalchemy.engine import URL, Connection from sqlalchemy.ext.asyncio import create_async_engine -from jentic_one.migrations.targets import DB_METADATA +from jentic_one.migrations.targets import DB_TARGETS from jentic_one.shared.config import DatabaseConfig, load_config from jentic_one.shared.db.backends import get_backend from jentic_one.shared.db.session import get_database_url @@ -53,7 +53,7 @@ def _on_version_apply( def _resolve_db_name() -> str: """Determine the target database from the Alembic config section name.""" section = config.config_ini_section - if section in DB_METADATA: + if section in DB_TARGETS: return section return "registry" @@ -111,7 +111,17 @@ def is_postgres() -> bool: def get_target_metadata() -> MetaData: """Return the metadata for the active migration target.""" - return DB_METADATA[_resolve_db_name()] + return DB_TARGETS[_resolve_db_name()].metadata + + +def get_version_table() -> str: + """Return the ``alembic_version`` table name for the active migration target. + + The built-in targets share the default ``alembic_version`` (scoped per-schema + by ``version_table_schema``); a target may use a distinct name to avoid a + version-tracking collision when it shares a schema with another target. + """ + return DB_TARGETS[_resolve_db_name()].version_table def _include_name( @@ -130,6 +140,19 @@ def _include_name( return True +def _include_object( + obj: Any, name: str | None, type_: str, reflected: bool, compare_to: Any +) -> bool: + """Default: include everything (each built-in target is single-schema). + + Overridable seam: a downstream ``env.py`` can replace this to treat certain + schemas as strictly read-only — returning ``False`` for objects whose schema + it does not own — so autogenerate never emits DROP/ALTER against tables it can + legitimately see (for cross-schema FK validation) but does not own. + """ + return True + + def run_migrations_offline() -> None: """Run migrations in 'offline' mode.""" url = get_url() @@ -139,9 +162,11 @@ def run_migrations_offline() -> None: target_metadata=get_target_metadata(), literal_binds=True, dialect_opts={"paramstyle": "named"}, + version_table=get_version_table(), version_table_schema=get_schema() if postgres else None, include_schemas=postgres, include_name=_include_name if postgres else None, + include_object=_include_object, render_as_batch=not postgres, on_version_apply=_on_version_apply, # Each migration owns its own transaction so that migrations using @@ -158,9 +183,11 @@ def do_run_migrations(connection: Connection) -> None: context.configure( connection=connection, target_metadata=get_target_metadata(), + version_table=get_version_table(), version_table_schema=get_schema() if postgres else None, include_schemas=postgres, include_name=_include_name if postgres else None, + include_object=_include_object, render_as_batch=not postgres, on_version_apply=_on_version_apply, # Each migration owns its own transaction so that migrations using diff --git a/src/jentic_one/migrations/run.py b/src/jentic_one/migrations/run.py index 79c4e152..4988c4ce 100644 --- a/src/jentic_one/migrations/run.py +++ b/src/jentic_one/migrations/run.py @@ -23,13 +23,16 @@ from alembic import command from alembic.config import Config -from jentic_one.migrations.targets import DB_METADATA - -VALID_DBS: tuple[str, ...] = tuple(DB_METADATA.keys()) +from jentic_one.migrations.targets import DB_TARGETS _MIGRATIONS_DIR = Path(__file__).resolve().parent +def _valid_dbs() -> tuple[str, ...]: + """Live target names (dynamic so targets registered post-import count).""" + return tuple(DB_TARGETS.keys()) + + def _build_config(db_name: str) -> Config: """Construct an in-memory Alembic config for a single database section.""" cfg = Config() @@ -42,10 +45,16 @@ def _build_config(db_name: str) -> Config: def upgrade(db_name: str, target: str = "head") -> None: """Apply migrations for a single database up to ``target``.""" - if db_name not in VALID_DBS: - raise ValueError(f"Unknown database {db_name!r}; expected one of {VALID_DBS}") - cfg = _build_config(db_name) - command.upgrade(cfg, target) + if db_name not in DB_TARGETS: + raise ValueError(f"Unknown database {db_name!r}; expected one of {_valid_dbs()}") + command.upgrade(_build_config(db_name), target) + + +def downgrade(db_name: str, target: str) -> None: + """Roll a single database back to ``target`` (e.g. ``"-1"`` or a revision/base).""" + if db_name not in DB_TARGETS: + raise ValueError(f"Unknown database {db_name!r}; expected one of {_valid_dbs()}") + command.downgrade(_build_config(db_name), target) def main(argv: list[str] | None = None) -> int: @@ -53,21 +62,40 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument( "--db", action="append", - choices=VALID_DBS, - help="Database to migrate (repeatable). Defaults to all databases.", + choices=_valid_dbs(), + help="Database to migrate (repeatable). Defaults to all, in dependency order.", + ) + parser.add_argument( + "--direction", + choices=("up", "down"), + default="up", + help="Migration direction (default: up).", ) parser.add_argument( "--target", - default="head", - help="Target revision (default: head).", + default=None, + help="Target revision. Default: 'head' (up) / '-1' (down). " + "The down default of '-1' is applied per --db, so a bare " + "'--db a --db b down' steps each database back one revision.", ) args = parser.parse_args(argv) - databases = args.db or list(VALID_DBS) - for db_name in databases: - print(f"==> Migrating {db_name} to {args.target}", flush=True) - upgrade(db_name, args.target) - print(f"==> {db_name} complete", flush=True) + order = args.db or list(_valid_dbs()) + if args.direction == "down": + # Rollback reverses registration order so a dependent schema tears down + # before the schema it FKs into. Critical for FK safety. + order = list(reversed(order)) + target = args.target or "-1" + for db_name in order: + print(f"==> Rolling back {db_name} to {target}", flush=True) + downgrade(db_name, target) + print(f"==> {db_name} rolled back", flush=True) + else: + target = args.target or "head" + for db_name in order: + print(f"==> Migrating {db_name} to {target}", flush=True) + upgrade(db_name, target) + print(f"==> {db_name} complete", flush=True) return 0 diff --git a/src/jentic_one/migrations/targets.py b/src/jentic_one/migrations/targets.py index 3c28f2a3..80dd54db 100644 --- a/src/jentic_one/migrations/targets.py +++ b/src/jentic_one/migrations/targets.py @@ -1,7 +1,20 @@ -"""Migration target metadata mapping for all databases.""" +"""Migration target registry for all databases. + +Replaces the old hardcoded ``DB_METADATA`` dict with an ordered registry of +:class:`MigrationTarget` records (name + metadata + version-table name) plus a +:func:`register_target` hook. The built-in ``registry``/``control``/``admin`` +targets register at import; a downstream package can import this module and +append its own isolated target via ``register_target(MigrationTarget(...))`` +without editing this file. + +Insertion order is the canonical UPGRADE order; rollback reverses it (see +``run.py``). This registry is the single source of truth for migration sequencing. +""" from __future__ import annotations +from dataclasses import dataclass + from sqlalchemy import MetaData import jentic_one.admin.core.schema # registers admin models on AdminBase.metadata @@ -9,8 +22,44 @@ import jentic_one.registry.core.schema # noqa: F401 # registers registry models on RegistryBase.metadata from jentic_one.shared.db.base import AdminBase, ControlBase, RegistryBase -DB_METADATA: dict[str, MetaData] = { - "registry": RegistryBase.metadata, - "control": ControlBase.metadata, - "admin": AdminBase.metadata, -} + +@dataclass(frozen=True, slots=True) +class MigrationTarget: + """A single migration target: its metadata and its version-table name. + + ``version_table`` names the per-target ``alembic_version`` table. Because all + targets share one Postgres instance (separated by schema), each must track its + own revision head in its own version table (in its own schema) — otherwise two + targets collide on a single ``alembic_version`` and clobber each other's head. + The built-in targets keep the default (``alembic_version``, already scoped + per-schema by ``env.py``'s ``version_table_schema``); the field only matters + when a target needs a distinct version table within a shared schema. + """ + + name: str + metadata: MetaData + version_table: str = "alembic_version" + + +#: Ordered registry of migration targets, keyed by name. Insertion order is the +#: canonical UPGRADE order (rollback reverses it — see run.py). +DB_TARGETS: dict[str, MigrationTarget] = {} + + +def register_target(target: MigrationTarget) -> None: + """Register a migration target. Idempotent for the same ``(name, target)``.""" + existing = DB_TARGETS.get(target.name) + if existing is not None and existing != target: + raise ValueError(f"Migration target {target.name!r} already registered") + DB_TARGETS[target.name] = target + + +register_target(MigrationTarget("registry", RegistryBase.metadata)) +register_target(MigrationTarget("control", ControlBase.metadata)) +register_target(MigrationTarget("admin", AdminBase.metadata)) + + +#: Backward-compatible name→metadata mapping. Derived from :data:`DB_TARGETS` so +#: any code (or test) still importing ``DB_METADATA`` keeps working. Prefer +#: ``DB_TARGETS`` in new code (it also carries ``version_table``). +DB_METADATA: dict[str, MetaData] = {name: t.metadata for name, t in DB_TARGETS.items()} diff --git a/src/jentic_one/shared/broker/__init__.py b/src/jentic_one/shared/broker/__init__.py index d12732a2..46ab529d 100644 --- a/src/jentic_one/shared/broker/__init__.py +++ b/src/jentic_one/shared/broker/__init__.py @@ -1,5 +1,16 @@ """Shared broker primitives: token resolution protocol and data types.""" +from jentic_one.shared.broker.execution import ( + ErrorOrigin, + ExecutionContext, + ExecutionOutcome, + RunnerRequest, + RunnerResult, + StreamingOutcome, + StreamingResult, + StreamingUpstreamRunner, + UpstreamRunner, +) from jentic_one.shared.broker.protocols import ( EgressPolicy, PluggableUpstreamRunner, @@ -11,15 +22,26 @@ UpstreamResult, Verb, ) +from jentic_one.shared.broker.schemas import ExecuteRequestContext __all__ = [ "EgressPolicy", + "ErrorOrigin", + "ExecuteRequestContext", + "ExecutionContext", + "ExecutionOutcome", "PluggableUpstreamRunner", "RunnerCapabilities", + "RunnerRequest", + "RunnerResult", + "StreamingOutcome", + "StreamingResult", + "StreamingUpstreamRunner", "Target", "TokenResolverProtocol", "ToolkitBindingCheckerProtocol", "UpstreamRequest", "UpstreamResult", + "UpstreamRunner", "Verb", ] diff --git a/src/jentic_one/shared/broker/broker.py b/src/jentic_one/shared/broker/broker.py new file mode 100644 index 00000000..34b180cb --- /dev/null +++ b/src/jentic_one/shared/broker/broker.py @@ -0,0 +1,73 @@ +"""Neutral Broker seam: the behavioral interface the data plane calls. + +Lives in ``shared/broker`` (the arch-neutral seam package) so the broker surface +and any downstream implementation depend on the Protocol, not the concrete +pipeline. The default implementation is ``DefaultBroker`` +(``broker/default_broker.py``); a downstream package can inject its own +implementation via ``AppContainer`` without importing broker internals. + +Two behavioral entry points, over the shared +:mod:`jentic_one.shared.broker.execution` / :mod:`jentic_one.shared.broker.schemas` +value objects: + +* ``execute`` — the buffered path (mirrors ``BrokerExecutionPipeline.execute``); + the sync router adapts the outcome to a ``Response`` and the async worker + persists the record. +* ``execute_streaming`` — the sync streaming passthrough (mirrors what + ``open_streaming_response`` needs); returns a ready ``Response`` streaming the + upstream body. Routing streaming through the seam means an injected broker's + controls are not bypassed on non-idempotent streaming traffic. + +Signature drift between an implementation and this contract is caught by +``BaseBrokerComplianceTest`` (``jentic_one.testing``); Protocols do not check +method signatures themselves. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import Protocol, runtime_checkable + +from fastapi import Response + +from jentic_one.shared.broker.execution import ( + ExecutionContext, + ExecutionOutcome, + RunnerRequest, + StreamingOutcome, + StreamingUpstreamRunner, +) +from jentic_one.shared.broker.schemas import ExecuteRequestContext + + +@runtime_checkable +class Broker(Protocol): + """Executes a resolved upstream call end-to-end (transport + post-response). + + The sync router awaits ``execute`` (buffered) or ``execute_streaming`` + (passthrough) and returns the ``Response``; the async worker runs ``execute`` + and persists the record. Neither re-implements the steps. An implementation + may wrap or replace the default (e.g. a wrapper delegating to a + ``DefaultBroker`` for the standard path) as long as it honors this contract. + + An injected broker **owns its own transport and resilience** (circuit + breaking, connection pooling, retries, bulkheads, timeouts): the built-in + resilience stack is applied by the runner that wraps the *default* broker, and + both callers use an injected ``Broker`` verbatim. To keep the built-in stack, + wrap a ``DefaultBroker`` and delegate to it. + """ + + async def execute( + self, request: RunnerRequest, context: ExecutionContext + ) -> ExecutionOutcome: ... + + async def execute_streaming( + self, + runner: StreamingUpstreamRunner, + request: RunnerRequest, + ctx_req: ExecuteRequestContext, + execution_id: str, + *, + transfer_deadline_s: float, + background_callback: Callable[[StreamingOutcome], Awaitable[None]] | None = None, + ) -> Response: ... diff --git a/src/jentic_one/shared/broker/execution.py b/src/jentic_one/shared/broker/execution.py new file mode 100644 index 00000000..e11ff104 --- /dev/null +++ b/src/jentic_one/shared/broker/execution.py @@ -0,0 +1,134 @@ +"""Transport-neutral broker execution value objects. + +These are the request/result/context/outcome value objects and runner protocols +that make up the public :class:`~jentic_one.shared.broker.broker.Broker` contract. +They live in ``shared/broker`` — not ``broker/`` — so both the broker surface and +any downstream implementation depend on the *same* value types without ``shared`` +importing ``broker`` (forbidden by ``tests/arch/test_module_boundaries.py``). + +The concrete broker pipeline (``broker/services/execution/pipeline.py``), the +runner adapters (``broker/adapters/runners/base.py``), and the streaming web edge +(``broker/web/streaming.py``) re-export these names, so existing broker-internal +call sites keep importing them from their old modules unchanged; this module is +the single definition. +""" + +from __future__ import annotations + +import time +from collections.abc import AsyncIterator +from contextlib import AbstractAsyncContextManager +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Protocol, runtime_checkable + +from jentic_one.shared.schemas import APIReference + + +class ErrorOrigin(StrEnum): + """Disambiguates *who* failed on an otherwise-identical 5xx/503.""" + + BROKER = "broker" + """The broker/system failed (overloaded, misconfigured, bad request shape).""" + + UPSTREAM = "upstream" + """The external vendor failed (down, rate-limited, returned an error).""" + + +@dataclass(frozen=True, slots=True) +class RunnerRequest: + """A transport-agnostic upstream request handed to a runner.""" + + method: str + url: str + headers: dict[str, str] = field(default_factory=dict) + body: bytes | None = None + timeout_s: float = 30.0 + + +@dataclass(frozen=True, slots=True) +class RunnerResult: + """The runner's response — the real upstream status/headers/body, verbatim.""" + + status_code: int + body: bytes + headers: dict[str, str] + content_type: str | None + duration_ms: int + + +@dataclass(frozen=True, slots=True) +class StreamingResult: + """A streamed upstream response: status/headers known up front, body lazy. + + Unlike :class:`RunnerResult` the body is **not** materialised — ``aiter`` is + the raw (still-compressed) upstream byte stream, consumed by the web edge to + feed a ``StreamingResponse``. It is only valid for the lifetime of the + ``stream()`` context manager that produced it; the upstream connection is + torn down when that context exits. + """ + + status_code: int + headers: dict[str, str] + content_type: str | None + aiter: AsyncIterator[bytes] + + +@dataclass(frozen=True, slots=True) +class ExecutionContext: + """Identity + discovery metadata threaded through the pipeline for one call.""" + + execution_id: str + toolkit_id: str | None + operation_id: str | None + api: APIReference | None + trace_id: str + + +@dataclass(frozen=True, slots=True) +class ExecutionOutcome: + """The immutable result a post-response stage may enrich (never the 2xx body).""" + + result: RunnerResult + context: ExecutionContext + error_origin: ErrorOrigin | None = None + + +@dataclass +class StreamingOutcome: + """Captures the final state of a streaming execution for persistence.""" + + execution_id: str + http_status: int + started_at_perf: float = field(default_factory=time.perf_counter) + duration_ms: int = 0 + error: str | None = None + bytes_transferred: int = 0 + + +@runtime_checkable +class UpstreamRunner(Protocol): + """Executes a single upstream request and returns its verbatim result. + + Decorators (retry/circuit/deadline/idempotency) implement this same protocol + and wrap a base runner, so the pipeline composes them without the handler + knowing which capabilities are active. + """ + + async def run(self, request: RunnerRequest) -> RunnerResult: ... + + +@runtime_checkable +class StreamingUpstreamRunner(UpstreamRunner, Protocol): + """An ``UpstreamRunner`` that can also stream the body without buffering (§08 E2.4). + + ``stream`` is an **async context manager**: it dispatches the request, yields + a :class:`StreamingResult` once the status/headers are in, and — critically — + holds the upstream ``httpx`` response open only for the duration of the + ``async with``. When the context exits (normal completion, size-cap/deadline + abort, or a client-disconnect ``CancelledError`` propagating out of the + body generator) the upstream stream is ``aclose()``d, releasing the pool slot + rather than leaking a zombie background drain. + """ + + def stream(self, request: RunnerRequest) -> AbstractAsyncContextManager[StreamingResult]: ... diff --git a/src/jentic_one/shared/broker/schemas.py b/src/jentic_one/shared/broker/schemas.py new file mode 100644 index 00000000..a29469e7 --- /dev/null +++ b/src/jentic_one/shared/broker/schemas.py @@ -0,0 +1,38 @@ +"""Shared broker request-context schema. + +``ExecuteRequestContext`` is part of the public :class:`~jentic_one.shared.broker.broker.Broker` +contract (the streaming entry point threads it through), so it lives in +``shared/broker`` — not ``broker/`` — allowing both the broker surface and any +downstream implementation to depend on the same type without ``shared`` +importing ``broker`` (forbidden by ``tests/arch/test_module_boundaries.py``). + +``broker/core/schemas.py`` re-exports this name, so existing broker-internal call +sites keep importing it from their old module unchanged; this module is the +single definition. +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel + + +class ExecuteRequestContext(BaseModel): + """Contextual metadata for a broker proxy request — discovery-driven. + + ``toolkit_id`` is no longer a required inbound header: it is derived (§03) or + supplied as an inbound disambiguator. ``operation_id`` / ``api_*`` come from + in-process discovery, not inbound ``Jentic-Api-*`` headers. + """ + + upstream_url: str + method: str + trace_id: str + toolkit_id: str | None = None + operation_id: str | None = None + api_vendor: str | None = None + api_name: str | None = None + api_version: str | None = None + prefer: str | None = None + pinned_revisions: dict[str, Any] | None = None diff --git a/src/jentic_one/shared/config.py b/src/jentic_one/shared/config.py index 70aa05d6..ace03c59 100644 --- a/src/jentic_one/shared/config.py +++ b/src/jentic_one/shared/config.py @@ -10,7 +10,16 @@ import structlog import yaml -from pydantic import BaseModel, BeforeValidator, Field, SecretStr, field_validator, model_validator +from pydantic import ( + BaseModel, + BeforeValidator, + ConfigDict, + Field, + SecretStr, + ValidationError, + field_validator, + model_validator, +) from jentic_one.shared.state.factory import StateBackendConfig @@ -693,19 +702,23 @@ class BrokerConfig(BaseModel): class SearchConfig(BaseModel): - """Lexical search configuration. - - Search is lexical (full-text / BM25) only: ingest builds an - ``operations.search_text`` projection and the query is matched against it. + """Search configuration. + + The built-in mode is "lexical" (BM25 on SQLite, native full-text on + PostgreSQL). ``search_mode`` is validated against the registered + SearchStrategy set at resolve time (``resolve_strategy``), so an unknown mode + fails loudly with the available modes for the active dialect rather than at + config load. Additional modes (e.g. "semantic", "vector") can be registered + via ``register_strategy`` without editing this schema. """ # Gate ingest-time construction of the lexical search_text projection. enabled: bool = True # Toggle query-time search independently of ingest-time indexing. search_enabled: bool = True - # Only "lexical" is supported in the open-source build (BM25 on SQLite, - # native full-text on PostgreSQL). - search_mode: Literal["lexical"] = "lexical" + # Search mode name; resolved against the SearchStrategy registry per dialect. + # "lexical" is the built-in mode. + search_mode: str = "lexical" class IngestConfig(BaseModel): @@ -784,6 +797,12 @@ def _instance_id_fits_column(cls, value: str | None) -> str | None: class AppConfig(BaseModel): """Top-level application configuration.""" + # Reject unknown top-level keys that are neither a known field nor a + # registered extension — surfaces misconfig loudly instead of dropping it. + # Registered extension sections are extracted by load_config() before + # validation, so they never reach this model as unknown keys. + model_config = ConfigDict(extra="forbid") + databases: DatabasesConfig services: ServicesConfig = Field(default_factory=ServicesConfig) worker: WorkerConfig = Field(default_factory=WorkerConfig) @@ -803,6 +822,50 @@ class AppConfig(BaseModel): telemetry: TelemetryConfig = Field(default_factory=TelemetryConfig) apps: list[str] = Field(default_factory=lambda: ["registry", "admin", "control", "auth"]) + # Validated extension sub-configs, keyed by their registered section name. + # Populated by load_config() from top-level keys matching the registry + # (see register_config). Empty unless a section has been registered. + extensions: dict[str, BaseModel] = Field(default_factory=dict) + + def extension(self, name: str) -> BaseModel | None: + """Return a registered extension config by section name (None if absent).""" + return self.extensions.get(name) + + +# --- Extension config registry ----------------------------------------------- +# A downstream package registers extra sub-config models at import time; by +# default the registry is empty. Keyed by the top-level YAML/env section name. +# load_config() pulls any matching top-level key out of the merged config and +# validates it with the registered model, storing the result in +# AppConfig.extensions[name]. +_CONFIG_EXTENSIONS: dict[str, type[BaseModel]] = {} + + +def register_config(name: str, model: type[BaseModel]) -> None: + """Register an extension sub-config model under a top-level config key. + + Idempotent for the same (name, model); raises on a conflicting re-register so + two extensions can't fight over one key. Call at import time (e.g. in a + registering package's __init__) before load_config() runs. + """ + # Collision guard: an extension key must not shadow a core AppConfig field + # (e.g. "broker", "search") nor the reserved "extensions" container itself — + # either would break the parser or silently override core config. + if name in AppConfig.model_fields or name == "extensions": + raise ConfigError( + f"Config extension name {name!r} collides with a core AppConfig field " + "or the reserved 'extensions' key" + ) + existing = _CONFIG_EXTENSIONS.get(name) + if existing is not None and existing is not model: + raise ConfigError(f"Config extension {name!r} already registered to {existing!r}") + _CONFIG_EXTENSIONS[name] = model + + +def registered_config_models() -> dict[str, type[BaseModel]]: + """Snapshot of the extension registry (for tests/introspection).""" + return dict(_CONFIG_EXTENSIONS) + def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: """Recursively merge override into base, preferring override values.""" @@ -868,6 +931,19 @@ def load_config(path: Path | None = None) -> AppConfig: if "apps" in merged and isinstance(merged["apps"], str): merged["apps"] = [item.strip() for item in merged["apps"].split(",") if item.strip()] + # Extract registered extension sections before validating the core model, so + # extra="forbid" accepts them and each is validated by its own model. + extensions: dict[str, BaseModel] = {} + for name, model in _CONFIG_EXTENSIONS.items(): + if name in merged: + raw = merged.pop(name) + try: + extensions[name] = model.model_validate(raw) + except ValidationError as e: + raise ConfigError(f"Invalid config for extension {name!r}: {e}") from e + if extensions: + merged["extensions"] = extensions + try: return AppConfig.model_validate(merged) except Exception as e: diff --git a/src/jentic_one/shared/events/__init__.py b/src/jentic_one/shared/events/__init__.py index 64e48be8..01af532f 100644 --- a/src/jentic_one/shared/events/__init__.py +++ b/src/jentic_one/shared/events/__init__.py @@ -10,7 +10,7 @@ from jentic_one.admin.repos.event_repo import EventRepository from jentic_one.shared.models.events import EVENT_TAGS, EventSeverity, EventTag, EventType -from jentic_one.shared.telemetry.events import TELEMETRY_EVENTS +from jentic_one.shared.telemetry.events import resolve_wire_name from jentic_one.shared.telemetry.sink import get_active_sink logger = structlog.get_logger(__name__) @@ -48,8 +48,9 @@ def _forward_to_telemetry(type: str, valid_tags: list[EventTag], actor_type: str if sink is None or not sink.enabled: return # Only forward events on the telemetry allowlist (internal-only events stay - # internal); the tag set has already been validated by the caller. - wire_name = TELEMETRY_EVENTS.get(type) + # internal); the tag set has already been validated by the caller. The + # resolver consults the built-in map first, then the runtime registry. + wire_name = resolve_wire_name(type) if wire_name is None: return sink.record(wire_name, valid_tags, actor_type) diff --git a/src/jentic_one/shared/jobs/protocols.py b/src/jentic_one/shared/jobs/protocols.py index a7b148b9..d5f49e4a 100644 --- a/src/jentic_one/shared/jobs/protocols.py +++ b/src/jentic_one/shared/jobs/protocols.py @@ -83,7 +83,7 @@ class UpstreamExecutor(Protocol): The async worker depends on this protocol — never on ``broker/`` — so the concrete broker adapter (``PipelineExecutor``, wrapping - ``run_execution(pipeline=default_pipeline(runner))``) can be dependency- + ``run_execution(broker=default_broker(runner))``) can be dependency- injected at worker startup. This is the "one pipeline, two callers" seam (§00 / §05 / §11 RN-0.3): the worker goes through the **same** composed runner (circuit breaker + per-host bulkhead + post-response enrichment) and diff --git a/src/jentic_one/shared/telemetry/events.py b/src/jentic_one/shared/telemetry/events.py index 737566bf..3797fcf0 100644 --- a/src/jentic_one/shared/telemetry/events.py +++ b/src/jentic_one/shared/telemetry/events.py @@ -12,6 +12,7 @@ from __future__ import annotations +import re from dataclasses import dataclass from datetime import datetime from enum import StrEnum @@ -89,6 +90,51 @@ class TelemetryEventName(StrEnum): } +# --- Runtime-registered events ----------------------------------------------- +# Built-in events stay in the closed TelemetryEventName enum + TELEMETRY_EVENTS +# map above. A downstream package can register extra (internal EventType value -> +# wire name) pairs here at import time. Wire names are validated plain strings; +# they never enter the closed enum, so strict typing of the built-in set is +# unaffected. The emit path resolves via resolve_wire_name(), which consults the +# built-in map first, then this registry. +_RUNTIME_TELEMETRY_EVENTS: dict[str, str] = {} + +# Wire-name syntax the ingest side accepts: lower_snake_case, matching the enum. +_WIRE_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*$") + + +def register_telemetry_event(event_type_key: str, wire_name: str) -> None: + """Register an extra event: internal EventType value -> wire name. + + Rejects collisions with the built-in enum/map and malformed wire names so a + typo can't silently disable forwarding. Idempotent for the same pair. Call at + import time (e.g. in a registering package's __init__). + """ + if not _WIRE_NAME_RE.match(wire_name): + raise ValueError(f"Invalid telemetry wire name {wire_name!r}") + if event_type_key in TELEMETRY_EVENTS: + raise ValueError(f"{event_type_key!r} is already a built-in telemetry event") + if wire_name in {e.value for e in TelemetryEventName}: + raise ValueError(f"Wire name {wire_name!r} collides with a built-in event") + existing = _RUNTIME_TELEMETRY_EVENTS.get(event_type_key) + if existing is not None and existing != wire_name: + raise ValueError(f"{event_type_key!r} already registered to {existing!r}") + _RUNTIME_TELEMETRY_EVENTS[event_type_key] = wire_name + + +def resolve_wire_name(event_type_key: str) -> str | None: + """Resolve an internal EventType value to a wire name, built-ins first. + + Returns None when the event is internal-only (not forwarded). This is the + single lookup the emit path uses instead of indexing TELEMETRY_EVENTS + directly, so built-in and registered events flow through one chokepoint. + """ + builtin = TELEMETRY_EVENTS.get(event_type_key) + if builtin is not None: + return builtin.value + return _RUNTIME_TELEMETRY_EVENTS.get(event_type_key) + + @dataclass(frozen=True, slots=True) class TelemetryEvent: """A single queued telemetry event. @@ -106,7 +152,7 @@ class TelemetryEvent: ``actor_type`` rather than by-convention. """ - name: TelemetryEventName + name: TelemetryEventName | str tags: tuple[EventTag, ...] ts: datetime actor_type: ActorType | None = None diff --git a/src/jentic_one/shared/telemetry/sink.py b/src/jentic_one/shared/telemetry/sink.py index 4500ad80..ded20b62 100644 --- a/src/jentic_one/shared/telemetry/sink.py +++ b/src/jentic_one/shared/telemetry/sink.py @@ -60,7 +60,7 @@ def queue(self) -> asyncio.Queue[TelemetryEvent]: def record( self, - name: TelemetryEventName, + name: TelemetryEventName | str, tags: Iterable[EventTag] = (), actor_type: str | None = None, ) -> None: diff --git a/src/jentic_one/shared/web/app_factory.py b/src/jentic_one/shared/web/app_factory.py index 841b16fb..e68499b9 100644 --- a/src/jentic_one/shared/web/app_factory.py +++ b/src/jentic_one/shared/web/app_factory.py @@ -34,6 +34,7 @@ from jentic_one.shared.telemetry.loop import TelemetryFlushLoop from jentic_one.shared.telemetry.sink import TelemetrySink, set_active_sink from jentic_one.shared.tracing import instrument_inbound_app +from jentic_one.shared.web.container import AppContainer from jentic_one.shared.web.openapi_meta import ( fastapi_metadata_kwargs, install_openapi_metadata, @@ -325,6 +326,7 @@ def create_surface_app( title: str, routers: Sequence[tuple[APIRouter, str, list[str]]], extra_lifespan: Callable[[FastAPI], AbstractAsyncContextManager[None]] | None = None, + container: AppContainer | None = None, ) -> FastAPI: """Create a standalone surface FastAPI app with observability wired. @@ -335,7 +337,12 @@ def create_surface_app( ``extra_lifespan`` is an optional surface-owned async context manager entered after ``ctx.startup()`` and exited before ``ctx.shutdown()`` — the broker uses it to open/close its shared outbound ``httpx.AsyncClient`` (§04). + + ``container`` is the DI seam: when omitted the default is used and behavior is + unchanged. A caller passes its own container to inject a ``Broker`` (stashed on + ``app.state``) and mount extra routers after the surface's own. """ + container = container or AppContainer.default(ctx) @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None]: @@ -373,8 +380,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: meta["title"] = title app = FastAPI(lifespan=lifespan, **meta) app.state.ctx = ctx + if container.broker is not None: + # Wire the injected broker into BOTH data-plane paths: the sync router + # reads app.state.broker, while the async worker builds its broker via + # app.state.broker_factory(runner). Without the factory the async path + # silently falls back to the default broker. + app.state.broker = container.broker + app.state.broker_factory = lambda _runner: container.broker for router, _prefix, tags in routers: app.include_router(router, tags=list(tags), responses=COMMON_ERROR_RESPONSES) + for extra_router, extra_prefix, extra_tags in container.extra_routers: + app.include_router( + extra_router, + prefix=extra_prefix, + tags=list(extra_tags), + responses=COMMON_ERROR_RESPONSES, + ) + for installer in container.extra_installers: + installer(app, ctx) app.add_middleware(RequestIDMiddleware) app.add_exception_handler(ProblemDetailException, problem_detail_exception_handler) # type: ignore[arg-type] attach_http_observability(app) @@ -382,8 +405,19 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: return app -def create_combined_app(ctx: Context, apps: list[str]) -> FastAPI: - """Build a root FastAPI that includes routers from all enabled surfaces.""" +def create_combined_app( + ctx: Context, + apps: list[str], + *, + container: AppContainer | None = None, +) -> FastAPI: + """Build a root FastAPI that includes routers from all enabled surfaces. + + ``container`` is the DI seam: when omitted, the default is used and behavior is + unchanged. A caller passes its own container to inject a ``Broker`` and mount + extra routers/installers after all built-in surfaces. + """ + container = container or AppContainer.default(ctx) structlog.get_logger(__name__).info("creating combined app", apps=apps) @asynccontextmanager @@ -403,6 +437,14 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: root = FastAPI(lifespan=lifespan, **fastapi_metadata_kwargs()) root.state.ctx = ctx + # Injected Broker (None by default → broker surface builds its default + # per request). Wire BOTH data-plane paths: the sync router reads + # root.state.broker; the async worker builds its broker via + # root.state.broker_factory(runner). Without the factory the async path + # silently falls back to the default broker. + if container.broker is not None: + root.state.broker = container.broker + root.state.broker_factory = lambda _runner: container.broker root.add_exception_handler(ProblemDetailException, problem_detail_exception_handler) # type: ignore[arg-type] @root.get( @@ -435,6 +477,15 @@ async def health() -> JSONResponse: # the reference it builds covers every included route. root.include_router(get_reference_router()) + # Extension point: injected routers/installers mount after all built-in + # surfaces (append-only; never shadows a built-in route). No-op by default. + for router, prefix, tags in container.extra_routers: + root.include_router( + router, prefix=prefix, tags=list(tags), responses=COMMON_ERROR_RESPONSES + ) + for installer in container.extra_installers: + installer(root, ctx) + root.add_middleware(RequestIDMiddleware) attach_http_observability(root) install_openapi_metadata(root) diff --git a/src/jentic_one/shared/web/container.py b/src/jentic_one/shared/web/container.py new file mode 100644 index 00000000..aa3cac89 --- /dev/null +++ b/src/jentic_one/shared/web/container.py @@ -0,0 +1,62 @@ +"""AppContainer: the dependency-injection seam for the app factories. + +Carries the pluggable pieces a caller may swap (Broker, extra routers/ +installers). The default container is constructed from a Context; a downstream +package can build its own and pass it to the factories. Keeping this in +``shared/web`` (not a surface) mirrors ``wiring.py``'s role as the composition +seam — it imports only ``shared/*`` + fastapi, so it introduces no +surface-boundary violation (``Broker`` lives in ``shared/broker``; +``DefaultBroker`` stays in ``broker/`` and is only referenced by the composition +root, never by ``shared/``). + +All fields default to the standard wiring, so passing only a Context reproduces +today's behavior; existing callers that pass just ``ctx`` are unaffected. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field + +from fastapi import APIRouter, FastAPI + +from jentic_one.shared.broker.broker import Broker +from jentic_one.shared.context import Context + +#: A mounted router spec: ``(router, prefix, tags)`` — same shape the surface +#: ``get_routers()`` functions return. ``tags`` is a ``Sequence[str]`` (not +#: ``list``) so it composes with FastAPI's invariant ``list[str | Enum]`` param +#: after a ``list(...)`` copy at the mount site. +RouterSpec = tuple[APIRouter, str, Sequence[str]] + +#: An installer run against the root app after all built-in surfaces are wired. +Installer = Callable[[FastAPI, Context], None] + + +@dataclass +class AppContainer: + """Injected dependencies for building an app. + + ``broker`` is the data-plane implementation (``DefaultBroker`` by default; + ``None`` here means "let the broker surface build its own default per + request"). ``extra_routers`` and ``extra_installers`` run against the root app + *after* the built-in surfaces are wired, so a caller mounts its + routers/handlers last and never shadows a built-in route. + + An injected ``broker`` owns its own transport + resilience (it opts out of the + built-in resilience stack) — see the ``Broker`` protocol docstring. + """ + + ctx: Context + broker: Broker | None = None + extra_routers: Sequence[RouterSpec] = field(default_factory=tuple) + extra_installers: Sequence[Installer] = field(default_factory=tuple) + + @classmethod + def default(cls, ctx: Context) -> AppContainer: + """Default container (no extra injection). + + ``broker`` stays ``None`` → the broker surface builds a ``DefaultBroker`` + per request via its ``broker_factory`` (see ``broker/services/execution``). + """ + return cls(ctx=ctx) diff --git a/src/jentic_one/testing/__init__.py b/src/jentic_one/testing/__init__.py new file mode 100644 index 00000000..11163a0f --- /dev/null +++ b/src/jentic_one/testing/__init__.py @@ -0,0 +1,21 @@ +"""Public testing seam: compliance base classes for extension points. + +A downstream package inherits these in its own test suite to prove its +``SearchStrategy`` / ``Broker`` implementations match the built-in contract — +both the method *names* (``runtime_checkable`` ``isinstance``) and the concrete +*signatures* (``inspect.signature``), which a bare Protocol never checks. + +Shipped in the wheel so ``import jentic_one.testing`` works for downstream repos. +""" + +from jentic_one.testing.compliance import ( + BaseBrokerComplianceTest, + BaseSearchStrategyComplianceTest, + assert_signature_matches, +) + +__all__ = [ + "BaseBrokerComplianceTest", + "BaseSearchStrategyComplianceTest", + "assert_signature_matches", +] diff --git a/src/jentic_one/testing/compliance.py b/src/jentic_one/testing/compliance.py new file mode 100644 index 00000000..a4e24dc2 --- /dev/null +++ b/src/jentic_one/testing/compliance.py @@ -0,0 +1,111 @@ +"""Signature-checking compliance bases for the extension-point seams. + +``runtime_checkable`` Protocols validate method *presence* only — an +implementation can drift in parameter names, defaults, or return type and still +pass ``isinstance``. These bases close that gap: each subclass points at its +implementation and inherits tests that assert both the ``isinstance`` conformance +*and* the exact ``inspect.signature`` of every seam method. + +Base classes are named ``Base*`` (not ``Test*``) so pytest does not collect them +directly; a subclass named ``Test*`` in the consuming test suite is what runs. +""" + +from __future__ import annotations + +import inspect + +from jentic_one.registry.repos.search.protocol import SearchStrategy +from jentic_one.shared.broker.broker import Broker + +#: Dialect names a backend can report from ``DatabaseBackend.dialect_name``. +#: Keep in sync with the backends in ``jentic_one.shared.db.backends`` — a +#: strategy whose ``dialect`` is not one of these can never be resolved by +#: ``resolve_strategy`` (it keys on ``backend.dialect_name``), so a plausible but +#: wrong value like ``"postgresql"`` would pass a bare ``isinstance(dialect, str)`` +#: check yet fail at runtime. This constant makes that class of drift a test failure. +KNOWN_BACKEND_DIALECTS: frozenset[str] = frozenset({"postgres", "sqlite"}) + + +def assert_signature_matches(impl: type, protocol: type, method: str) -> None: + """Assert ``impl.method`` has the same signature as ``protocol.method``. + + Closes the ``runtime_checkable`` gap (which only checks method presence). + Ignores ``self`` and compares parameter names, kinds, defaults, and + annotations, plus the return annotation. + """ + proto_sig = inspect.signature(getattr(protocol, method)) + impl_sig = inspect.signature(getattr(impl, method)) + proto_params = list(proto_sig.parameters.values())[1:] # drop self + impl_params = list(impl_sig.parameters.values())[1:] + assert impl_params == proto_params, ( + f"{impl.__name__}.{method} parameters diverge from " + f"{protocol.__name__}.{method}: {impl_sig} != {proto_sig}" + ) + assert impl_sig.return_annotation == proto_sig.return_annotation, ( + f"{impl.__name__}.{method} return type diverges from " + f"{protocol.__name__}.{method}: {impl_sig.return_annotation!r} != " + f"{proto_sig.return_annotation!r}" + ) + + +class BaseSearchStrategyComplianceTest: + """Subclass and set ``strategy_cls`` to prove a ``SearchStrategy`` conforms. + + The subclass sets ``strategy_cls`` to the zero-arg-constructible strategy + class, e.g.:: + + class TestMyStrategyCompliance(BaseSearchStrategyComplianceTest): + strategy_cls = MyStrategy + """ + + #: The strategy class under test. Subclasses override this. + strategy_cls: type + + def test_is_search_strategy(self) -> None: + assert isinstance(self.strategy_cls(), SearchStrategy) + + def test_has_required_attrs(self) -> None: + # Instantiate first: `name`/`dialect` may be an instance @property (which + # evaluates to a property object on the class, not a str). Asserting on an + # instance handles both plain class attrs and properties. + instance = self.strategy_cls() + assert isinstance(instance.name, str) + assert isinstance(instance.dialect, str) + + def test_dialect_is_resolvable(self) -> None: + # A string `dialect` is not enough: it must match a real backend's + # `dialect_name` or `resolve_strategy((dialect, mode))` never finds this + # strategy and raises SearchUnsupportedError at runtime. Catch the + # plausible-but-wrong value (e.g. "postgresql" vs "postgres") here. + instance = self.strategy_cls() + assert instance.dialect in KNOWN_BACKEND_DIALECTS, ( + f"{self.strategy_cls.__name__}.dialect={instance.dialect!r} is not a known " + f"backend dialect {sorted(KNOWN_BACKEND_DIALECTS)}; resolve_strategy would " + "never find this strategy." + ) + + def test_search_operations_signature(self) -> None: + assert_signature_matches(self.strategy_cls, SearchStrategy, "search_operations") + + +class BaseBrokerComplianceTest: + """Subclass and override ``broker_factory`` to prove a ``Broker`` conforms. + + ``broker_factory`` returns a ready ``Broker`` instance, e.g.:: + + class TestMyBrokerCompliance(BaseBrokerComplianceTest): + def broker_factory(self) -> Broker: + return MyBroker(...) + """ + + def broker_factory(self) -> Broker: + raise NotImplementedError("Subclass must override broker_factory()") + + def test_is_broker(self) -> None: + assert isinstance(self.broker_factory(), Broker) + + def test_execute_signature(self) -> None: + assert_signature_matches(type(self.broker_factory()), Broker, "execute") + + def test_execute_streaming_signature(self) -> None: + assert_signature_matches(type(self.broker_factory()), Broker, "execute_streaming") diff --git a/src/jentic_one/wiring.py b/src/jentic_one/wiring.py index d7261cb2..2c16ad2b 100644 --- a/src/jentic_one/wiring.py +++ b/src/jentic_one/wiring.py @@ -23,6 +23,7 @@ from jentic_one.shared.broker.protocols import ResolveResult, RevisionPinResult from jentic_one.shared.context import Context from jentic_one.shared.db.session import DatabaseSession +from jentic_one.shared.web.container import AppContainer class InProcessRegistryResolver: @@ -66,3 +67,13 @@ async def resolve_revision_pin( def install_broker_registry_resolver(app: FastAPI, ctx: Context) -> None: """Inject the in-process registry resolver onto the broker app state.""" app.state.broker_registry_resolver = InProcessRegistryResolver(ctx.registry_db) + + +def build_default_container(ctx: Context) -> AppContainer: + """Assemble the default ``AppContainer`` (no extra injection). + + The composition root's factory for the DI seam. A downstream package can + provide its own ``build_container`` that starts here and adds its ``Broker`` / + extra routers, then calls the same app factories with the resulting container. + """ + return AppContainer.default(ctx) diff --git a/tests/arch/_rules_facts.py b/tests/arch/_rules_facts.py new file mode 100644 index 00000000..0b036e45 --- /dev/null +++ b/tests/arch/_rules_facts.py @@ -0,0 +1,84 @@ +"""Resolve and load machine-readable rules facts published by the rules repo. + +The ``jentic-one-rules`` repo is the source of truth for the parameterizable +enforcement facts (e.g. the ORM conventions in ``orm.facts.yaml``). This module +resolves that file with a robust lookup order and validates its +``schema_version`` so the arch tests can read the facts instead of hard-coding +them. A vendored copy is committed here so a standalone clone (with no external +rules repo mounted) still self-enforces. + +Lookup order (first hit wins): + +1. ``$JENTIC_RULES_DIR/rules/backend/`` — explicit override (a mounted or + local read-only clone; see ``scripts/rules-clone.sh``). +2. ``/.rules/rules/backend/`` — in-repo clone auto-detected with no + env var. ``scripts/rules-clone.sh`` clones here by default; ``.rules/`` is + gitignored so its private content can never be committed. +3. ``/..//rules/backend/`` — sibling checkout, where + ```` is ``$JENTIC_RULES_SIBLING_NAME`` (default ``jentic-one-rules``). +4. ``tests/arch/vendored/`` — committed fallback so a standalone clone + (no rules repo mounted) still self-enforces. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any + +import yaml + +_ARCH_DIR = Path(__file__).resolve().parent +_REPO_ROOT = _ARCH_DIR.parent.parent +VENDORED_DIR = _ARCH_DIR / "vendored" + +#: In-repo (gitignored) mount path that ``scripts/rules-clone.sh`` clones into. +#: Auto-detected so a dev who runs that script needs no env var at all. +_IN_REPO_MOUNT = _REPO_ROOT / ".rules" + +#: Default name of the sibling rules checkout beside this repo. Overridable via +#: ``$JENTIC_RULES_SIBLING_NAME`` so the private name isn't the sole load-bearing +#: reference (forks / mirrors can point elsewhere). +_DEFAULT_SIBLING_NAME = "jentic-one-rules" + + +def candidate_paths(name: str) -> list[Path]: + """Return the ordered lookup paths for a rules facts file named *name*.""" + rel = Path("rules") / "backend" / name + paths: list[Path] = [] + env = os.environ.get("JENTIC_RULES_DIR") + if env: + paths.append(Path(env) / rel) + paths.append(_IN_REPO_MOUNT / rel) + sibling = os.environ.get("JENTIC_RULES_SIBLING_NAME") or _DEFAULT_SIBLING_NAME + paths.append(_REPO_ROOT.parent / sibling / rel) + paths.append(VENDORED_DIR / name) + return paths + + +def mounted_facts_path(name: str) -> Path | None: + """First existing non-vendored candidate, or None if only the vendored copy exists.""" + for p in candidate_paths(name): + if p.is_file() and p.parent != VENDORED_DIR: + return p + return None + + +def load_facts(name: str, *, expected_schema: str) -> dict[str, Any]: + """Load and schema-validate the first available facts file named *name*.""" + for p in candidate_paths(name): + if p.is_file(): + data: dict[str, Any] = yaml.safe_load(p.read_text(encoding="utf-8")) + got = data.get("schema_version") + if got != expected_schema: + raise ValueError(f"{p}: schema_version {got!r} != expected {expected_schema!r}") + return data + raise FileNotFoundError( + f"Could not locate rules facts {name!r}. Tried: " + + ", ".join(str(p) for p in candidate_paths(name)) + ) + + +def orm_facts() -> dict[str, Any]: + """Load the ORM enforcement facts (``orm.facts.yaml``).""" + return load_facts("orm.facts.yaml", expected_schema="jentic.orm-facts/v1") diff --git a/tests/arch/test_orm_conventions.py b/tests/arch/test_orm_conventions.py index 014c7beb..01065093 100644 --- a/tests/arch/test_orm_conventions.py +++ b/tests/arch/test_orm_conventions.py @@ -2,6 +2,9 @@ All models inheriting from RegistryBase, ControlBase, or AdminBase must follow project naming and structural conventions. + +Enforced facts are sourced from the rules repo's ``orm.facts.yaml`` (see +``_rules_facts.py``); do not hard-code them here. """ from __future__ import annotations @@ -12,14 +15,22 @@ import pytest +from ._rules_facts import orm_facts from .conftest import SRC_ROOT, python_files_in -VALID_BASES = frozenset({"RegistryBase", "ControlBase", "AdminBase"}) +_FACTS = orm_facts() + +VALID_BASES = frozenset(_FACTS["valid_bases"]) # Mixins that provide audit columns (created_at, updated_at, created_by) to any # model that inherits them. A model gaining created_at via one of these mixins # satisfies the created_at convention without declaring the column directly. -AUDIT_MIXINS = frozenset({"AuditableMixin"}) +AUDIT_MIXINS = frozenset(_FACTS["audit_mixins"]) + +_TABLENAME_FACTS = _FACTS["tablename"] +_SNAKE_CASE_REQUIRED = bool(_TABLENAME_FACTS["snake_case"]) +_PLURAL_SUFFIXES = tuple(_TABLENAME_FACTS["plural_suffixes"]) +_REQUIRED_COLUMNS = tuple(_FACTS["required_columns"]) def _get_base_name(node: ast.expr) -> str | None: @@ -92,13 +103,13 @@ def _check_model(filepath: Path, cls_node: ast.ClassDef) -> list[str]: f"(all ORM models must set __tablename__ explicitly)" ) else: - if not _is_snake_case(tablename): + if _SNAKE_CASE_REQUIRED and not _is_snake_case(tablename): violations.append( f"{loc} — class {class_name} has __tablename__ = '{tablename}' " f"(table names must be snake_case, e.g. 'api_specs')" ) - if not tablename.endswith("s") and not tablename.endswith("data"): + if not tablename.endswith(_PLURAL_SUFFIXES): violations.append( f"{loc} — class {class_name} has __tablename__ = '{tablename}' " f"(table names should be plural, e.g. 'jobs' not 'job')" @@ -106,14 +117,18 @@ def _check_model(filepath: Path, cls_node: ast.ClassDef) -> list[str]: columns = _get_column_names(cls_node) - if "id" not in columns: + if "id" in _REQUIRED_COLUMNS and "id" not in columns: violations.append( f"{loc} — class {class_name} missing 'id' column " f"(all tables must have a primary key named 'id')" ) # created_at may be declared directly or inherited from an audit mixin. - if "created_at" not in columns and not _has_base(cls_node, AUDIT_MIXINS): + if ( + "created_at" in _REQUIRED_COLUMNS + and "created_at" not in columns + and not _has_base(cls_node, AUDIT_MIXINS) + ): violations.append( f"{loc} — class {class_name} missing 'created_at' column " f"(all tables must have a created_at timestamp)" @@ -180,38 +195,7 @@ def test_orm_models_have_created_at() -> None: assert not violations, "ORM models missing 'created_at' timestamp:\n" + "\n".join(violations) -EXEMPT_FROM_KSUID = frozenset( - { - "apis", - "api_revisions", - "operation_url_indexes", - "security_schemes", - "security_scheme_flows", - "spec_files", - "servers", - "server_variables", - "operations", - "oauth_client_credentials", - "basic_credentials", - "token_value_credentials", - # Structurally single-row cache of the upstream catalog manifest. Its id is - # a fixed sentinel (CatalogSnapshot.SINGLETON_ID), not a per-entity ksuid — - # that constant key is what makes a concurrent refresh collide instead of - # appending a second snapshot row. - "catalog_snapshots", - # Structurally single-row first-run lock. Its id is a fixed sentinel - # (SETUP_SENTINEL_ID), not a per-entity ksuid — that constant key is what - # makes a second concurrent first-admin attempt collide on the primary key - # instead of inserting a second row. A ksuid default would defeat the lock. - "setup_sentinels", - # Structurally single-row holder of the opaque telemetry instance id. Its - # id is a fixed sentinel (INSTANCE_IDENTITY_ID == "singleton"), not a - # per-entity ksuid — that constant key is what makes a concurrent seed on - # a second replica collide on the primary key (the loser reads the row), - # which is exactly the once-per-instance dedupe. - "instance_identities", - } -) +EXEMPT_FROM_KSUID = frozenset(_FACTS["ksuid_exempt_tables"]) def _check_ksuid_default(filepath: Path, cls_node: ast.ClassDef) -> list[str]: @@ -267,6 +251,8 @@ def _is_generate_ksuid_call(node: ast.expr) -> bool: @pytest.mark.arch def test_orm_id_uses_ksuid() -> None: """Non-exempt ORM models must use generate_ksuid for their id server_default.""" + if _FACTS["primary_key"] != "ksuid": + pytest.skip(f"primary_key strategy is {_FACTS['primary_key']!r}, not 'ksuid'") violations: list[str] = [] for py_file in python_files_in(SRC_ROOT): if py_file.is_relative_to(SRC_ROOT / "migrations"): diff --git a/tests/arch/test_rules_facts_vendored.py b/tests/arch/test_rules_facts_vendored.py new file mode 100644 index 00000000..88d2f029 --- /dev/null +++ b/tests/arch/test_rules_facts_vendored.py @@ -0,0 +1,119 @@ +"""Guard the vendored rules-facts fallback: OSS-safety and upstream drift. + +Two independent guards protect the committed ``tests/arch/vendored/orm.facts.yaml``: + +1. ``test_vendored_orm_facts_is_oss_safe`` (ALWAYS runs): the vendored file may + contain ONLY facts that are meaningful in this public repo. It fails if an + unexpected top-level section, an unknown ORM base, or a table name that does + not exist in this repo's models sneaks in — so a re-vendor can never leak + content that only makes sense in a downstream (non-public) repo. + +2. ``test_vendored_orm_facts_matches_mounted_source`` (runs only when the + external rules repo is mounted): ensures the vendored copy has not drifted + from the upstream source of truth. +""" + +from __future__ import annotations + +import ast + +import pytest +import yaml + +from ._rules_facts import VENDORED_DIR, mounted_facts_path, orm_facts +from .conftest import SRC_ROOT, python_files_in + +# Top-level keys the vendored ORM facts file is allowed to carry. Anything else +# (e.g. a downstream-only section) must not appear in the public repo. +_ALLOWED_FACT_KEYS = frozenset( + { + "schema_version", + "primary_key", + "valid_bases", + "audit_mixins", + "tablename", + "required_columns", + "ksuid_exempt_tables", + } +) + + +def _oss_declarative_bases() -> frozenset[str]: + """Return the ORM declarative base class names defined in this repo.""" + base_file = SRC_ROOT / "shared" / "db" / "base.py" + tree = ast.parse(base_file.read_text(encoding="utf-8")) + bases: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef) and node.name.endswith("Base"): + bases.add(node.name) + return frozenset(bases) + + +def _oss_tablenames() -> frozenset[str]: + """Return every ``__tablename__`` string literal declared in this repo's models.""" + names: set[str] = set() + for py_file in python_files_in(SRC_ROOT): + try: + tree = ast.parse(py_file.read_text(encoding="utf-8")) + except SyntaxError: + continue + for node in ast.walk(tree): + if not isinstance(node, ast.Assign): + continue + for target in node.targets: + if ( + isinstance(target, ast.Name) + and target.id == "__tablename__" + and isinstance(node.value, ast.Constant) + and isinstance(node.value.value, str) + ): + names.add(node.value.value) + return frozenset(names) + + +@pytest.mark.arch +def test_vendored_orm_facts_is_oss_safe() -> None: + """The vendored facts must reference only symbols that exist in this repo. + + This is the primary leak guard: it turns "the vendored file must not carry + downstream-only content" from a review promise into an enforced test that + runs in this repo's CI, with or without the external rules repo mounted. + """ + facts = orm_facts() + + unexpected_keys = set(facts) - _ALLOWED_FACT_KEYS + assert not unexpected_keys, ( + f"Vendored orm.facts.yaml has unexpected top-level keys {sorted(unexpected_keys)}. " + f"Only {sorted(_ALLOWED_FACT_KEYS)} are allowed in the public repo — re-vendor the " + "OSS-applicable subset only." + ) + + oss_bases = _oss_declarative_bases() + unknown_bases = set(facts.get("valid_bases", [])) - oss_bases + assert not unknown_bases, ( + f"Vendored orm.facts.yaml lists valid_bases {sorted(unknown_bases)} that are not " + f"defined in this repo (known: {sorted(oss_bases)}). A base that only exists downstream " + "must not appear in the public vendored facts." + ) + + oss_tables = _oss_tablenames() + unknown_tables = set(facts.get("ksuid_exempt_tables", [])) - oss_tables + assert not unknown_tables, ( + f"Vendored orm.facts.yaml exempts tables {sorted(unknown_tables)} that have no " + "__tablename__ in this repo's models. A table that only exists downstream must not " + "appear in the public vendored facts." + ) + + +@pytest.mark.arch +def test_vendored_orm_facts_matches_mounted_source() -> None: + mounted = mounted_facts_path("orm.facts.yaml") + if mounted is None: + pytest.skip("rules repo not mounted; vendored copy is authoritative") + vendored = VENDORED_DIR / "orm.facts.yaml" + assert yaml.safe_load(vendored.read_text(encoding="utf-8")) == yaml.safe_load( + mounted.read_text(encoding="utf-8") + ), ( + "Vendored orm.facts.yaml drifted from the mounted rules repo. " + "Re-vendor: copy the mounted file over tests/arch/vendored/orm.facts.yaml." + ) diff --git a/tests/arch/vendored/README.md b/tests/arch/vendored/README.md new file mode 100644 index 00000000..737d70d4 --- /dev/null +++ b/tests/arch/vendored/README.md @@ -0,0 +1,46 @@ +# Vendored rules facts + +This directory holds a **committed subset** of the machine-readable enforcement +facts published by the `jentic-one-rules` repo. Today that is a single file: + +- `orm.facts.yaml` — the ORM conventions (`valid_bases`, required columns, + tablename rules, KSUID-exempt tables) that `tests/arch/test_orm_conventions.py` + reads instead of hard-coding. + +## Why it's vendored + +The arch tests resolve facts in priority order (see `tests/arch/_rules_facts.py`): + +1. `$JENTIC_RULES_DIR` — an explicitly-pointed mounted/cloned rules repo. +2. `.rules/` — an in-repo clone auto-detected with no env var (what + `scripts/rules-clone.sh` creates; gitignored). +3. a sibling `../jentic-one-rules` checkout. +4. **this vendored copy** — the fallback so a standalone clone with no access to + the rules repo still self-enforces its architecture. + +## The no-leak contract + +This is a **public** repo. The vendored file may contain **only** facts that are +meaningful in this repo. Concretely: + +- **OSS-applicable facts only.** No section, base, or table name that exists only + in a downstream (non-public) tier may be copied here. +- This is **enforced**, not a promise: `test_vendored_orm_facts_is_oss_safe` + (in `tests/arch/test_rules_facts_vendored.py`, always run) fails if the file + gains an unexpected top-level key, a `valid_bases` entry not defined in + `src/jentic_one/shared/db/base.py`, or a `ksuid_exempt_tables` entry with no + matching `__tablename__` in this repo's models. + +## Re-vendoring (when upstream facts change) + +1. Get the upstream rules repo locally: run `scripts/rules-clone.sh` (clones into + the auto-detected `.rules/`), or set `JENTIC_RULES_DIR=/path/to/jentic-one-rules`. +2. Run the guards: + `uv run pytest tests/arch/test_rules_facts_vendored.py tests/arch/test_orm_conventions.py`. + The drift guard (`test_vendored_orm_facts_matches_mounted_source`) will fail + if the vendored copy is stale. +3. Copy the upstream file over the vendored one: + `cp "$JENTIC_RULES_DIR/rules/backend/orm.facts.yaml" tests/arch/vendored/orm.facts.yaml`. +4. Re-run the guards. `test_vendored_orm_facts_is_oss_safe` must still pass — if + it fails, the upstream file carries content that is not OSS-safe and must not + be vendored as-is. diff --git a/tests/arch/vendored/orm.facts.yaml b/tests/arch/vendored/orm.facts.yaml new file mode 100644 index 00000000..a3251339 --- /dev/null +++ b/tests/arch/vendored/orm.facts.yaml @@ -0,0 +1,25 @@ +# Vendored fallback of jentic-one-rules/rules/backend/orm.facts.yaml — do not hand-edit except to track upstream (test_rules_facts_vendored.py guards it when the rules repo is mounted). +schema_version: jentic.orm-facts/v1 +primary_key: ksuid +valid_bases: [RegistryBase, ControlBase, AdminBase] +audit_mixins: [AuditableMixin] +tablename: + snake_case: true + plural_suffixes: [s, data] +required_columns: [id, created_at] +ksuid_exempt_tables: + - apis + - api_revisions + - operation_url_indexes + - security_schemes + - security_scheme_flows + - spec_files + - servers + - server_variables + - operations + - oauth_client_credentials + - basic_credentials + - token_value_credentials + - catalog_snapshots + - setup_sentinels + - instance_identities diff --git a/tests/unit/broker/test_circuit_event.py b/tests/unit/broker/test_circuit_event.py index 966ad445..57955286 100644 --- a/tests/unit/broker/test_circuit_event.py +++ b/tests/unit/broker/test_circuit_event.py @@ -11,7 +11,7 @@ from jentic_one.broker.core.schemas import ExecuteRequestContext from jentic_one.broker.services.execution.service import ( _circuit_event_last_emitted, - default_pipeline, + default_broker, run_execution, ) @@ -52,7 +52,7 @@ def _clear_dedup() -> None: async def test_circuit_open_emits_event() -> None: """When CircuitOpenError is raised, emit_event should be called.""" session = AsyncMock() - pipeline = default_pipeline(_CircuitOpenRunner()) + broker = default_broker(_CircuitOpenRunner()) with ( patch( @@ -70,7 +70,7 @@ async def test_circuit_open_emits_event() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc123", actor_type="agent", ) @@ -89,7 +89,7 @@ async def test_circuit_open_emits_event() -> None: async def test_circuit_open_dedup_suppresses_second_emit() -> None: """A second CircuitOpenError for the same host within the cooldown should not emit.""" session = AsyncMock() - pipeline = default_pipeline(_CircuitOpenRunner()) + broker = default_broker(_CircuitOpenRunner()) with ( patch( @@ -108,7 +108,7 @@ async def test_circuit_open_dedup_suppresses_second_emit() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc123", actor_type="agent", ) @@ -135,14 +135,14 @@ async def test_circuit_open_different_hosts_emit_independently() -> None: ) as mock_emit, ): for host in ("alpha.example.com", "beta.example.com"): - pipeline = default_pipeline(_CircuitOpenRunner()) + broker = default_broker(_CircuitOpenRunner()) with pytest.raises(CircuitOpenError): await run_execution( _ctx_req(host=host), body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc123", actor_type="agent", ) diff --git a/tests/unit/broker/test_execution_events.py b/tests/unit/broker/test_execution_events.py index 295eabc2..7ea68a09 100644 --- a/tests/unit/broker/test_execution_events.py +++ b/tests/unit/broker/test_execution_events.py @@ -11,7 +11,7 @@ from jentic_one.broker.core.exceptions import BrokerError from jentic_one.broker.core.schemas import ExecuteRequestContext from jentic_one.broker.services.execution.service import ( - default_pipeline, + default_broker, persist_streaming_execution, run_execution, ) @@ -70,7 +70,7 @@ def _ctx_req() -> ExecuteRequestContext: async def test_run_execution_emits_completed_event() -> None: """Successful execution emits EXECUTION_COMPLETED.""" session = AsyncMock() - pipeline = default_pipeline(_SuccessRunner()) + broker = default_broker(_SuccessRunner()) with ( patch( @@ -87,7 +87,7 @@ async def test_run_execution_emits_completed_event() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc", actor_type="agent", ) @@ -104,7 +104,7 @@ async def test_run_execution_emits_completed_event() -> None: async def test_run_execution_emits_failed_event_on_broker_error() -> None: """BrokerError during execution emits EXECUTION_FAILED before re-raising.""" session = AsyncMock() - pipeline = default_pipeline(_FailRunner()) + broker = default_broker(_FailRunner()) with ( patch( @@ -122,7 +122,7 @@ async def test_run_execution_emits_failed_event_on_broker_error() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc", actor_type="agent", ) @@ -253,7 +253,7 @@ async def test_run_execution_tags_failed_event_with_thirdparty_auth( same-timestamp events, so a second event would skew the funnel. """ session = AsyncMock() - pipeline = default_pipeline(_StatusRunner(status_code)) + broker = default_broker(_StatusRunner(status_code)) with ( patch( @@ -270,7 +270,7 @@ async def test_run_execution_tags_failed_event_with_thirdparty_auth( body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc", actor_type="agent", ) @@ -290,7 +290,7 @@ async def test_run_execution_tags_failed_event_with_thirdparty_auth( async def test_run_execution_no_auth_tag_on_success() -> None: """A 2xx upstream response emits no tagged failure event.""" session = AsyncMock() - pipeline = default_pipeline(_SuccessRunner()) + broker = default_broker(_SuccessRunner()) with ( patch( @@ -307,7 +307,7 @@ async def test_run_execution_no_auth_tag_on_success() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc", actor_type="agent", ) @@ -448,7 +448,7 @@ async def test_persist_streaming_execution_no_auth_tag_without_vendor() -> None: async def test_run_execution_no_auth_tag_without_vendor() -> None: """A vendorless call (no credential path) emits an untagged failure on 401.""" session = AsyncMock() - pipeline = default_pipeline(_StatusRunner(401)) + broker = default_broker(_StatusRunner(401)) ctx_req = ExecuteRequestContext( upstream_url="https://api.example.com/v1/test", method="GET", @@ -477,7 +477,7 @@ async def test_run_execution_no_auth_tag_without_vendor() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc", actor_type="agent", ) diff --git a/tests/unit/broker/test_execution_observability.py b/tests/unit/broker/test_execution_observability.py index 06a813b7..42303cb5 100644 --- a/tests/unit/broker/test_execution_observability.py +++ b/tests/unit/broker/test_execution_observability.py @@ -13,7 +13,7 @@ from jentic_one.broker.adapters.runners.base import RunnerRequest, RunnerResult, UpstreamRunner from jentic_one.broker.core.schemas import ExecuteRequestContext -from jentic_one.broker.services.execution.service import default_pipeline, run_execution +from jentic_one.broker.services.execution.service import default_broker, run_execution class _StubRunner(UpstreamRunner): @@ -56,7 +56,7 @@ def span_exporter() -> Iterator[InMemorySpanExporter]: async def test_broker_execute_span_created(span_exporter: InMemorySpanExporter) -> None: """The broker.execute span is created with expected attributes.""" session = AsyncMock() - pipeline = default_pipeline(_StubRunner()) + broker = default_broker(_StubRunner()) with patch( "jentic_one.broker.services.execution.service.record_execution", new_callable=AsyncMock @@ -66,7 +66,7 @@ async def test_broker_execute_span_created(span_exporter: InMemorySpanExporter) body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc123", actor_type="agent", ) @@ -86,7 +86,7 @@ async def test_broker_execute_span_created(span_exporter: InMemorySpanExporter) async def test_execution_started_and_finished_log_events() -> None: """Structured log events execution_started and execution_finished fire.""" session = AsyncMock() - pipeline = default_pipeline(_StubRunner()) + broker = default_broker(_StubRunner()) factory = structlog.testing.CapturingLoggerFactory() @@ -105,7 +105,7 @@ async def test_execution_started_and_finished_log_events() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc123", actor_type="agent", ) diff --git a/tests/unit/broker/test_execution_service.py b/tests/unit/broker/test_execution_service.py index c9f11aa0..93be8888 100644 --- a/tests/unit/broker/test_execution_service.py +++ b/tests/unit/broker/test_execution_service.py @@ -8,7 +8,7 @@ from jentic_one.broker.adapters.runners.base import RunnerRequest, RunnerResult, UpstreamRunner from jentic_one.broker.core.schemas import ExecuteRequestContext -from jentic_one.broker.services.execution.service import default_pipeline, run_execution +from jentic_one.broker.services.execution.service import default_broker, run_execution class _StubRunner(UpstreamRunner): @@ -53,7 +53,7 @@ def _session_mock() -> AsyncMock: async def test_actor_fields_forwarded_to_record_execution() -> None: """actor_id and actor_type passed to run_execution reach record_execution.""" session = _session_mock() - pipeline = default_pipeline(_StubRunner()) + broker = default_broker(_StubRunner()) with patch( "jentic_one.broker.services.execution.service.record_execution", new_callable=AsyncMock @@ -64,7 +64,7 @@ async def test_actor_fields_forwarded_to_record_execution() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="agt_abc123", actor_type="agent", ) @@ -79,7 +79,7 @@ async def test_actor_fields_forwarded_to_record_execution() -> None: async def test_actor_fields_are_required() -> None: """actor_id and actor_type are required parameters for run_execution.""" session = _session_mock() - pipeline = default_pipeline(_StubRunner()) + broker = default_broker(_StubRunner()) with patch( "jentic_one.broker.services.execution.service.record_execution", new_callable=AsyncMock @@ -90,7 +90,7 @@ async def test_actor_fields_are_required() -> None: body=None, headers=None, session=session, - pipeline=pipeline, + broker=broker, actor_id="usr_default", actor_type="user", ) diff --git a/tests/unit/broker/test_injected_broker_seam.py b/tests/unit/broker/test_injected_broker_seam.py new file mode 100644 index 00000000..53e4b58f --- /dev/null +++ b/tests/unit/broker/test_injected_broker_seam.py @@ -0,0 +1,218 @@ +"""Injected-broker seam: proves an injected ``Broker`` reaches both callers. + +The value of the DI seam is "one pipeline, two callers": a downstream package +injects a single :class:`Broker` and it must be honored by **both** the sync +router (``broker/web/routers/execute.py``) and the async worker +(:class:`PipelineExecutor`) — not just one of them. + +This exercises the real selection code on both sides with one shared spy broker: + +- **Wiring**: an :class:`AppContainer` with a broker stashes it on + ``app.state.broker`` for both ``create_surface_app`` and ``create_combined_app`` + (the source the sync router's ``_handle`` reads). +- **Sync path**: ``_resolve_broker`` (the helper ``_handle`` uses) prefers + ``app.state.broker`` over the per-request ``broker_factory`` default. +- **Async path**: ``PipelineExecutor`` built with an injected ``broker_factory`` + runs the injected broker through ``run_execution`` (verified end-to-end via a + real in-memory execution against an ``AsyncMock`` session — the session is a + plain mock, DB internals are never mocked per ``tests/arch/test_no_db_mocking``). + +The spy is a compliant :class:`Broker` (``jentic_one.testing`` bases guard the +signature elsewhere); here we only assert *it was the one invoked*. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import Response + +from jentic_one.broker.adapters.runners.base import RunnerRequest as BrokerRunnerRequest +from jentic_one.broker.adapters.runners.base import ( + RunnerResult, + StreamingUpstreamRunner, + UpstreamRunner, +) +from jentic_one.broker.adapters.runners.registry import RunnerRegistry +from jentic_one.broker.services.execution.executor import PipelineExecutor +from jentic_one.broker.web.routers.execute import _resolve_broker +from jentic_one.shared.broker.broker import Broker +from jentic_one.shared.broker.execution import ( + ExecutionContext, + ExecutionOutcome, + RunnerRequest, + StreamingOutcome, +) +from jentic_one.shared.broker.execution import RunnerResult as SharedRunnerResult +from jentic_one.shared.broker.schemas import ExecuteRequestContext +from jentic_one.shared.config import AppConfig +from jentic_one.shared.context import Context +from jentic_one.shared.jobs.protocols import UpstreamExecRequest +from jentic_one.shared.web.app_factory import create_combined_app, create_surface_app +from jentic_one.shared.web.container import AppContainer + + +class _SpyBroker: + """A minimal compliant :class:`Broker` that records whether it was called.""" + + def __init__(self) -> None: + self.calls: list[tuple[RunnerRequest, ExecutionContext]] = [] + + async def execute(self, request: RunnerRequest, context: ExecutionContext) -> ExecutionOutcome: + self.calls.append((request, context)) + return ExecutionOutcome( + result=SharedRunnerResult( + status_code=200, + body=b"spy", + headers={}, + content_type="text/plain", + duration_ms=1, + ), + context=context, + ) + + async def execute_streaming( + self, + runner: StreamingUpstreamRunner, + request: RunnerRequest, + ctx_req: ExecuteRequestContext, + execution_id: str, + *, + transfer_deadline_s: float, + background_callback: Callable[[StreamingOutcome], Awaitable[None]] | None = None, + ) -> Response: # pragma: no cover - selection tests never invoke it + return Response(content=b"spy-stream", status_code=200) + + +class _NoopRunner(UpstreamRunner): + """A runner that must never be reached when an injected broker owns transport.""" + + async def run(self, request: BrokerRunnerRequest) -> RunnerResult: # pragma: no cover + raise AssertionError("injected broker must own transport; runner.run must not be called") + + +@pytest.fixture() +def ctx(sample_config_dict: dict[str, Any]) -> Context: + return Context(AppConfig.model_validate(sample_config_dict)) + + +def test_spy_broker_is_a_broker() -> None: + """Sanity: the spy honors the Broker protocol (isinstance seam).""" + assert isinstance(_SpyBroker(), Broker) + + +def test_container_stashes_injected_broker_on_combined_app(ctx: Context) -> None: + broker = _SpyBroker() + container = AppContainer(ctx=ctx, broker=broker) + app = create_combined_app(ctx, ["control"], container=container) + assert app.state.broker is broker + + +def test_container_stashes_injected_broker_on_surface_app(ctx: Context) -> None: + broker = _SpyBroker() + container = AppContainer(ctx=ctx, broker=broker) + app = create_surface_app( + ctx, + title="test-surface", + routers=[], + container=container, + ) + assert app.state.broker is broker + + +def test_default_container_leaves_broker_unset(ctx: Context) -> None: + """No injection → the sync router falls back to its per-request factory.""" + app = create_combined_app(ctx, ["control"]) + assert getattr(app.state, "broker", None) is None + + +def test_sync_path_selection_prefers_injected_broker() -> None: + """The real selection helper ``_handle`` uses: injected instance wins over factory. + + Calls ``broker/web/routers/execute.py::_resolve_broker`` directly (not a + copy) so this test tracks the production logic — an injected + ``app.state.broker`` is used verbatim; only its absence falls back to + ``broker_factory(runner)``. + """ + spy = _SpyBroker() + default_factory_called = False + + def _factory(_runner: UpstreamRunner) -> Broker: + nonlocal default_factory_called + default_factory_called = True + return _SpyBroker() + + request = MagicMock() + request.app.state.broker = spy + request.app.state.broker_factory = _factory + + assert _resolve_broker(request, _NoopRunner()) is spy + assert default_factory_called is False + + +def test_sync_path_selection_falls_back_to_factory_when_unset() -> None: + """With ``broker=None`` the per-request factory is used (fallback branch).""" + made = _SpyBroker() + factory_called = False + + def _factory(_runner: UpstreamRunner) -> Broker: + nonlocal factory_called + factory_called = True + return made + + request = MagicMock() + request.app.state.broker = None + request.app.state.broker_factory = _factory + + assert _resolve_broker(request, _NoopRunner()) is made + assert factory_called is True + + +@pytest.mark.asyncio +async def test_async_path_invokes_injected_broker() -> None: + """PipelineExecutor built with an injected factory runs the injected broker. + + Uses the real ``run_execution`` path (no monkeypatching of the service) so the + test proves the injected broker is actually invoked, not merely stored. The + session is a plain AsyncMock stand-in (DB internals are never mocked). + """ + spy = _SpyBroker() + + def _broker_factory(_runner: UpstreamRunner) -> Broker: + return spy + + registry = RunnerRegistry() + registry.register(["http", "https"], _NoopRunner(), required=True) + executor = PipelineExecutor(registry, broker_factory=_broker_factory) + + session = AsyncMock() + session.add = MagicMock() + + request = UpstreamExecRequest( + method="GET", + url="https://api.example.com/v1/things", + headers={}, + body=None, + timeout_s=30.0, + metadata={ + "execution_id": "exec_test0000000000000000", + "trace_id": "a" * 32, + "toolkit_id": "tk_test000000000000000000", + "operation_id": "getThing", + "api_vendor": "example", + "api_name": "api", + "api_version": "1.0.0", + "actor_id": "agt_abc123", + "actor_type": "agent", + "origin": "agent", + }, + ) + + result = await executor.execute(request, session=session) + + assert len(spy.calls) == 1, "injected broker must be invoked on the async path" + assert result.status_code == 200 + assert result.body == b"spy" diff --git a/tests/unit/broker/test_repeated_failure_emit.py b/tests/unit/broker/test_repeated_failure_emit.py index 72f921e7..52f9b483 100644 --- a/tests/unit/broker/test_repeated_failure_emit.py +++ b/tests/unit/broker/test_repeated_failure_emit.py @@ -36,7 +36,7 @@ from jentic_one.broker.core.exceptions import BrokerError from jentic_one.broker.core.schemas import ExecuteRequestContext from jentic_one.broker.services.execution.service import ( - default_pipeline, + default_broker, persist_streaming_execution, run_execution, ) @@ -163,7 +163,7 @@ async def test_run_execution_emits_repeated_failure_on_broker_error(session: Asy body=None, headers=None, session=session, - pipeline=default_pipeline(_FailRunner()), + broker=default_broker(_FailRunner()), actor_id=_ACTOR, actor_type="agent", security_config=config, @@ -189,7 +189,7 @@ async def test_run_execution_no_repeated_failure_without_security_config( body=None, headers=None, session=session, - pipeline=default_pipeline(_FailRunner()), + broker=default_broker(_FailRunner()), actor_id=_ACTOR, actor_type="agent", ) diff --git a/tests/unit/shared/telemetry/test_event_registry.py b/tests/unit/shared/telemetry/test_event_registry.py new file mode 100644 index 00000000..742ca390 --- /dev/null +++ b/tests/unit/shared/telemetry/test_event_registry.py @@ -0,0 +1,88 @@ +"""Tests for the telemetry-event extension seam. + +Covers ``register_telemetry_event`` + ``resolve_wire_name`` — the seam that lets +a downstream package forward extra events without editing the closed +``TelemetryEventName`` enum: + +- wire-name syntax validation (``lower_snake_case``), +- collision rejection against the built-in enum and the built-in event map, +- idempotency for the same pair, conflict on a re-register, +- ``resolve_wire_name`` precedence (built-ins first, then runtime registry, + ``None`` for internal-only events). + +Each test snapshots/restores the process-global runtime registry so a +registration never leaks into another test. +""" + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest + +from jentic_one.shared.telemetry import events as events_mod +from jentic_one.shared.telemetry.events import ( + TELEMETRY_EVENTS, + TelemetryEventName, + register_telemetry_event, + resolve_wire_name, +) + + +@pytest.fixture(autouse=True) +def _isolate_runtime_registry() -> Iterator[None]: + snapshot = dict(events_mod._RUNTIME_TELEMETRY_EVENTS) + try: + yield + finally: + events_mod._RUNTIME_TELEMETRY_EVENTS.clear() + events_mod._RUNTIME_TELEMETRY_EVENTS.update(snapshot) + + +def test_register_and_resolve_runtime_event() -> None: + register_telemetry_event("my_ext.thing_happened", "thing_happened") + assert resolve_wire_name("my_ext.thing_happened") == "thing_happened" + + +def test_resolve_wire_name_prefers_builtin() -> None: + """A built-in EventType resolves via the closed map, never the runtime one.""" + # PBAC_DENIED is a built-in mapping. + key = next(iter(TELEMETRY_EVENTS)) + assert resolve_wire_name(key) == TELEMETRY_EVENTS[key].value + + +def test_resolve_wire_name_none_for_internal_only() -> None: + assert resolve_wire_name("some.internal_only_event") is None + + +@pytest.mark.parametrize( + "bad_wire_name", + ["CamelCase", "with-dash", "1leading_digit", "has space", ""], +) +def test_register_rejects_malformed_wire_name(bad_wire_name: str) -> None: + with pytest.raises(ValueError, match="Invalid telemetry wire name"): + register_telemetry_event("my_ext.evt", bad_wire_name) + + +def test_register_rejects_builtin_event_type_key() -> None: + builtin_key = next(iter(TELEMETRY_EVENTS)) + with pytest.raises(ValueError, match="already a built-in telemetry event"): + register_telemetry_event(builtin_key, "some_new_wire_name") + + +def test_register_rejects_wire_name_colliding_with_builtin_enum() -> None: + builtin_wire = TelemetryEventName.BROKER_EXECUTION.value + with pytest.raises(ValueError, match="collides with a built-in event"): + register_telemetry_event("my_ext.custom", builtin_wire) + + +def test_register_is_idempotent_for_same_pair() -> None: + register_telemetry_event("my_ext.evt", "custom_event") + register_telemetry_event("my_ext.evt", "custom_event") # no raise + assert resolve_wire_name("my_ext.evt") == "custom_event" + + +def test_register_rejects_conflicting_reregister() -> None: + register_telemetry_event("my_ext.evt", "custom_event") + with pytest.raises(ValueError, match="already registered"): + register_telemetry_event("my_ext.evt", "different_wire_name") diff --git a/tests/unit/test_config_extensions.py b/tests/unit/test_config_extensions.py new file mode 100644 index 00000000..42526fa9 --- /dev/null +++ b/tests/unit/test_config_extensions.py @@ -0,0 +1,156 @@ +"""Tests for the config extension seam (register_config + AppConfig.extension). + +Covers the registrable extension sub-config mechanics that let a downstream +package add top-level config sections without editing the core schema: + +- ``register_config`` (registration + idempotency + collision guards), +- ``registered_config_models`` (registry snapshot), +- ``AppConfig.extension`` (typed lookup by section name), +- ``load_config`` extraction of a registered section (so ``extra="forbid"`` on + the core model accepts it) + rejection of *unregistered* top-level keys. + +Each test snapshots and restores the process-global ``_CONFIG_EXTENSIONS`` so it +never leaks a registration into another test. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +import pytest +import yaml +from pydantic import BaseModel, ValidationError + +from jentic_one.shared import config as config_mod +from jentic_one.shared.config import ( + AppConfig, + ConfigError, + load_config, + register_config, + registered_config_models, +) + + +class _ExtensionModel(BaseModel): + """A stand-in downstream extension sub-config.""" + + enabled: bool = False + label: str = "default" + + +class _OtherExtensionModel(BaseModel): + threshold: int = 10 + + +@pytest.fixture(autouse=True) +def _isolate_config_registry() -> Iterator[None]: + """Snapshot/restore the global extension registry around each test.""" + snapshot = dict(config_mod._CONFIG_EXTENSIONS) + try: + yield + finally: + config_mod._CONFIG_EXTENSIONS.clear() + config_mod._CONFIG_EXTENSIONS.update(snapshot) + + +def _minimal_config() -> dict[str, Any]: + return { + "databases": { + "registry": {"name": "reg"}, + "admin": {"name": "admin"}, + "control": {"name": "ctrl"}, + } + } + + +def test_register_config_adds_to_registry() -> None: + register_config("my_ext", _ExtensionModel) + assert registered_config_models()["my_ext"] is _ExtensionModel + + +def test_registered_config_models_is_a_copy() -> None: + """The snapshot must not be the live registry (mutating it can't corrupt state).""" + register_config("my_ext", _ExtensionModel) + snapshot = registered_config_models() + snapshot["injected"] = _OtherExtensionModel + assert "injected" not in registered_config_models() + + +def test_register_config_is_idempotent_for_same_pair() -> None: + register_config("my_ext", _ExtensionModel) + register_config("my_ext", _ExtensionModel) # no raise + assert registered_config_models()["my_ext"] is _ExtensionModel + + +def test_register_config_rejects_conflicting_reregister() -> None: + register_config("my_ext", _ExtensionModel) + with pytest.raises(ConfigError, match="already registered"): + register_config("my_ext", _OtherExtensionModel) + + +@pytest.mark.parametrize("core_field", ["broker", "search", "databases", "telemetry"]) +def test_register_config_rejects_core_field_names(core_field: str) -> None: + """An extension must not hijack a core AppConfig field.""" + assert core_field in AppConfig.model_fields + with pytest.raises(ConfigError, match="collides with a core AppConfig field"): + register_config(core_field, _ExtensionModel) + + +def test_register_config_rejects_reserved_extensions_key() -> None: + """The reserved ``extensions`` container itself cannot be shadowed.""" + with pytest.raises(ConfigError, match="reserved 'extensions' key"): + register_config("extensions", _ExtensionModel) + + +def test_appconfig_extension_returns_none_when_absent() -> None: + cfg = AppConfig.model_validate(_minimal_config()) + assert cfg.extension("my_ext") is None + + +def test_load_config_extracts_and_validates_registered_section(tmp_path: Path) -> None: + register_config("my_ext", _ExtensionModel) + data = _minimal_config() + data["my_ext"] = {"enabled": True, "label": "prod"} + path = tmp_path / "cfg.yaml" + path.write_text(yaml.dump(data)) + + cfg = load_config(path) + + ext = cfg.extension("my_ext") + assert isinstance(ext, _ExtensionModel) + assert ext.enabled is True + assert ext.label == "prod" + + +def test_load_config_invalid_extension_section_raises(tmp_path: Path) -> None: + register_config("my_ext", _ExtensionModel) + data = _minimal_config() + # ``label`` is a str; a nested mapping is the wrong type and fails validation + # inside the extension model (surfaced as a ConfigError, not a raw pydantic one). + data["my_ext"] = {"label": {"nested": "wrong-type"}} + path = tmp_path / "cfg.yaml" + path.write_text(yaml.dump(data)) + + with pytest.raises(ConfigError, match="Invalid config for extension 'my_ext'"): + load_config(path) + + +def test_load_config_rejects_unregistered_top_level_key(tmp_path: Path) -> None: + """A top-level key that is neither core nor registered fails loudly (extra=forbid).""" + data = _minimal_config() + data["totally_unknown_section"] = {"foo": "bar"} + path = tmp_path / "cfg.yaml" + path.write_text(yaml.dump(data)) + + with pytest.raises(ConfigError, match="validation failed"): + load_config(path) + + +def test_appconfig_forbids_unknown_top_level_field() -> None: + """The core model itself rejects unknown keys (the breaking-change guard).""" + data = _minimal_config() + data["typo_section"] = {"x": 1} + with pytest.raises(ValidationError, match=r"typo_section|extra"): + AppConfig.model_validate(data) diff --git a/tests/unit/test_migrations.py b/tests/unit/test_migrations.py index 21d71a14..eb66d93c 100644 --- a/tests/unit/test_migrations.py +++ b/tests/unit/test_migrations.py @@ -8,7 +8,12 @@ import pytest from sqlalchemy import MetaData -from jentic_one.migrations.targets import DB_METADATA +from jentic_one.migrations.targets import ( + DB_METADATA, + DB_TARGETS, + MigrationTarget, + register_target, +) from jentic_one.shared.db.base import AdminBase, ControlBase, RegistryBase @@ -52,6 +57,36 @@ def test_all_databases_covered() -> None: assert set(DB_METADATA.keys()) == {"registry", "control", "admin"} +def test_db_targets_cover_oss_surfaces() -> None: + assert set(DB_TARGETS.keys()) == {"registry", "control", "admin"} + + +def test_db_targets_carry_metadata_and_default_version_table() -> None: + for name, base in ( + ("registry", RegistryBase), + ("control", ControlBase), + ("admin", AdminBase), + ): + target = DB_TARGETS[name] + assert target.metadata is base.metadata + assert target.version_table == "alembic_version" + + +def test_db_metadata_shim_matches_targets() -> None: + assert {name: t.metadata for name, t in DB_TARGETS.items()} == DB_METADATA + + +def test_register_target_is_idempotent_for_same_target() -> None: + # Re-registering an identical target is a no-op (safe for repeat imports). + register_target(DB_TARGETS["registry"]) + assert set(DB_TARGETS.keys()) == {"registry", "control", "admin"} + + +def test_register_target_rejects_conflicting_redefinition() -> None: + with pytest.raises(ValueError, match="already registered"): + register_target(MigrationTarget("registry", ControlBase.metadata)) + + @pytest.mark.parametrize("db_name", ["registry", "control", "admin"]) def test_versions_directory_exists(db_name: str) -> None: versions_dir = ( diff --git a/tests/unit/testing/test_compliance_oss.py b/tests/unit/testing/test_compliance_oss.py new file mode 100644 index 00000000..0e2f541b --- /dev/null +++ b/tests/unit/testing/test_compliance_oss.py @@ -0,0 +1,57 @@ +"""Self-test for the ``jentic_one.testing`` compliance harness. + +Runs the shipped compliance bases against the built-in default implementations so +CI guarantees the harness itself is correct (and that the built-in SearchStrategy +/ Broker never drift from their seams). + +This repo forbids ``class Test*`` in test files (``tests/arch/test_no_test_classes``), +so instead of subclassing the bases with ``Test*`` names (the shape a consuming +test suite uses), we instantiate underscore-prefixed subclasses and drive their +inherited harness methods from plain parametrized functions. This exercises the +exact same harness code paths (``assert_signature_matches`` + ``isinstance``). +""" + +from __future__ import annotations + +import pytest + +from jentic_one.broker.adapters.runners.base import RunnerRequest, RunnerResult, UpstreamRunner +from jentic_one.broker.default_broker import DefaultBroker +from jentic_one.broker.services.execution.pipeline import BrokerExecutionPipeline +from jentic_one.registry.repos.search.postgres_lexical import PostgresLexicalStrategy +from jentic_one.registry.repos.search.sqlite_lexical import SqliteLexicalStrategy +from jentic_one.shared.broker.broker import Broker +from jentic_one.testing import BaseBrokerComplianceTest, BaseSearchStrategyComplianceTest + + +class _NoopRunner(UpstreamRunner): + async def run(self, request: RunnerRequest) -> RunnerResult: # pragma: no cover - unused + return RunnerResult(status_code=200, body=b"", headers={}, content_type=None, duration_ms=0) + + +class _SqliteCompliance(BaseSearchStrategyComplianceTest): + strategy_cls = SqliteLexicalStrategy + + +class _PostgresCompliance(BaseSearchStrategyComplianceTest): + strategy_cls = PostgresLexicalStrategy + + +class _DefaultBrokerCompliance(BaseBrokerComplianceTest): + def broker_factory(self) -> Broker: + return DefaultBroker(BrokerExecutionPipeline(_NoopRunner())) + + +@pytest.mark.parametrize("compliance", [_SqliteCompliance(), _PostgresCompliance()]) +def test_oss_search_strategies_comply(compliance: BaseSearchStrategyComplianceTest) -> None: + compliance.test_is_search_strategy() + compliance.test_has_required_attrs() + compliance.test_dialect_is_resolvable() + compliance.test_search_operations_signature() + + +def test_oss_default_broker_complies() -> None: + compliance = _DefaultBrokerCompliance() + compliance.test_is_broker() + compliance.test_execute_signature() + compliance.test_execute_streaming_signature() diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 0335938d..d5c9a5c7 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -77,6 +77,11 @@ function buildRoutes(): RouteObject[] { // Basename index: the authenticated app shell home (`/app`). path: '/', element: , + // `moduleRoutes` is the append-only route registry; it is + // spread here (not inlined) so a consumer can compose + // `[...moduleRoutes, ...extraRoutes]` at this single point + // without editing the registry. Order is load-bearing — + // placeholders stay last so a real module route wins. children: [dashboardIndexRoute, ...moduleRoutes, ...placeholderRoutes], }, ], diff --git a/ui/src/mocks/handlers.ts b/ui/src/mocks/handlers.ts index a1c6baba..fe33496d 100644 --- a/ui/src/mocks/handlers.ts +++ b/ui/src/mocks/handlers.ts @@ -147,6 +147,12 @@ export const handlers = [ // here keeps the Monitor page working in mocked dev when no other module's // handler claimed the path. ...monitorHandlers, + // Extensibility seam: `handlers` is exported (not module-private) so a + // consumer can compose `[...handlers, ...extraHandlers]`. MSW is + // FIRST-MATCH-WINS, so a consumer must append at a DELIBERATE position — + // appending after this array lets these fixtures answer first (safest + // default); to override a path here, a consumer must register its handler + // BEFORE this one in the composed array. ]; /** diff --git a/ui/vite.config.ts b/ui/vite.config.ts index d01796e2..1f91fdd6 100644 --- a/ui/vite.config.ts +++ b/ui/vite.config.ts @@ -103,5 +103,14 @@ export default defineConfig({ server: { host: '0.0.0.0', proxy: backendProxy, + // Extensibility seam: this config intentionally sets NO `server.fs` + // restriction. OSS uses the plain `@` alias for its own source (`ui/src`). + // A downstream host that consumes this SPA as a shared module graph must + // alias `@` -> this repo's `ui/src` (so OSS's own `@/` imports resolve + // here, not into the host tree) and use a DISTINCT prefix for its own + // code (e.g. `@ent/`) — it must NOT claim `@` for itself. The host adds a + // matching `server.fs.allow` entry in its own vite config, not here. + // Leaving `server.fs` unrestricted here avoids pre-emptively blocking that + // cross-package mount; do not hardcode `server.fs` in a way that would. }, });