diff --git a/.env.example b/.env.example index da762d51c..55336d75a 100644 --- a/.env.example +++ b/.env.example @@ -70,7 +70,12 @@ REFLEXIO_MAX_INTERACTION_CONTENT_TOKENS=512 # Concurrent durable extraction turns per process; acquired before a user lease. # options: positive integer (optional; defaults to publish concurrency, currently 4) REFLEXIO_PUBLISH_LEARNING_WORKERS=4 -# Durable discovery timing is code-owned: 2-second polling, 300-second renewable lease. +# Pickup latency versus discovery read load. +# options: finite positive seconds within the platform wait range (default 2) +REFLEXIO_DURABLE_LEARNING_POLL_SECONDS=2 +# Renewable background claim lifetime and worker-loss recovery. +# options: positive integral seconds within epoch/heartbeat headroom (default 300) +REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS=300 # Inactivity delay in seconds before a quiet session is evaluated by agent-success # evaluation. (optional; default 600 = 10 minutes) GROUP_EVALUATION_DELAY_SECONDS=600 diff --git a/README.md b/README.md index df177d89a..31a1be6ca 100644 --- a/README.md +++ b/README.md @@ -437,6 +437,14 @@ Client (SDK / CLI / HTTP API) └─ UnifiedSearchService → Profiles + user/agent playbooks ``` +The import-light [inference launch resolver](reflexio/inference_launch.py) gives +providers and launchers one endpoint and process-ownership contract. An explicit +service URL attaches to an existing inference process; otherwise the launcher +owns a separate child. See [developer.md](developer.md) for migration and runtime details. +The shared legacy boolean parser lives in [env_utils.py](reflexio/env_utils.py) +so launchers can reuse it without importing server startup. The existing +`reflexio.server.env_utils.env_truthy` import remains a compatibility export. + See [developer.md](developer.md) for project structure, supported LLM providers, and development setup. ## Documentation diff --git a/developer.md b/developer.md index a425c12e1..b322c590f 100644 --- a/developer.md +++ b/developer.md @@ -803,24 +803,64 @@ instruction. Unprepared extraction keeps its turn-reference contract. Strict source visibility validation, bounded corrective repair, and evidence-carrier limits are unchanged. -### Durable extraction timing - -`reflexio/durable_learning_policy.py` owns background discovery cadence (2 seconds) -and renewable user-claim duration (300 seconds). Library/server scheduling and -enterprise failure reporting share that policy. `REFLEXIO_DURABLE_LEARNING_POLL_SECONDS` -and `REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS` are retired: unset/blank or former -default values remain accepted as ignored compatibility inputs; custom or invalid -values fail startup. Remove those entries before upgrading a deployment that -customized them. Worker capacity remains independently configurable through -`REFLEXIO_PUBLISH_LEARNING_WORKERS`; claim fencing and heartbeat renewal are unchanged. +### Supabase storage schema compatibility + +`StorageConfigSupabase` accepts legacy `read_url`/`read_key` fields only to discard +them during normalization; they are absent from canonical output and the schema. +Enterprise Supabase reads use native SQL, so these HTTP reader fields did not +select a replica. `StorageConfigPostgres.read_db_url` and reader pool settings +remain supported for actual native replica routing. + +### Stored configuration validation + +Persisted reads use `validate_stored_config`: removed unknown fields are ignored +and missing fields use schema defaults, but invalid recognized values raise +`StoredConfigurationError` from `reflexio.config_migration`. Existing +`ConfigurationMigrationError` subclasses remain within this family. Local-file +loading and CLI storage discovery propagate these errors rather than creating or +writing a default policy. Correct the saved configuration before retrying. +Errors and adapter logs name only top-level schema settings; supplied values and +dynamic dictionary keys are omitted. API writes still use ordinary Pydantic +validation. Missing-file, unreadable JSON/decryption and explicit legacy-disk +handling are separate existing contracts, not changed by this validation rule. +### Durable extraction timing -Housekeeping timing and governance deletion batches live in +`reflexio/durable_learning_policy.py` resolves background discovery timing once +before worker construction. `REFLEXIO_DURABLE_LEARNING_POLL_SECONDS` controls +pickup latency and discovery read load (default 2 seconds); +`REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS` controls renewable background claim +lifetime and worker-loss recovery (default 300 seconds). These are independent +supported deployment inputs again: the previous retirement fixed their values +without removing their operational purpose. Valid custom values now take effect; +review complete resolved deployment settings when upgrading from the retirement. + +Blank values select defaults. Poll seconds must be finite and positive within +`threading.TIMEOUT_MAX`; lease seconds must be positive integral seconds within +signed-64-bit epoch and heartbeat wait headroom. Integral decimal spellings remain +accepted. Invalid values fail startup without disclosing supplied values. +`resolve_discovery_timing(mapping)` and `validate_discovery_timing_env(mapping)` +use the supplied mapping exclusively, even when empty. Scheduler and enterprise +failure reporting capture the same policy at construction; restart to change it. + +Worker capacity remains independently configurable through +`REFLEXIO_PUBLISH_LEARNING_WORKERS`. Claim fencing and heartbeat renewal are +unchanged. Explicit/manual user leases and pending-run claims retain separate +lifetimes and are not controlled by these background discovery settings. + + +Housekeeping cadence and lineage timing migration checks live in `reflexio/housekeeping_policy.py`. The former `lineage_gc.poll_interval_seconds` -and `governance_retention.audit_events_delete_batch_limit` settings accept only -legacy defaults (86400 and 500) during migration and are absent from canonical -configuration. Custom values fail validation; review them before upgrading. -Retention ages, grace windows and explicit holds remain supported. +accepts only its legacy default (86400) and is absent from canonical configuration; +custom lineage cadence still requires explicit migration. In contrast, +`governance_retention.audit_events_delete_batch_limit` is supported again: a +positive integer up to PostgreSQL int32, default 500. Enterprise audit cleanup +applies the configured limit separately to audit events and erasure tombstones +in the existing transaction. The earlier retirement did not eliminate this +throughput/transaction-size choice. Stored custom values are preserved; invalid +values fail loading without resetting other saved policy. Retention enablement, +ages, grace windows and holds remain independent. OSS accepts this enterprise +policy without executing audit-event cleanup. ### Session outcome cutovers @@ -829,3 +869,21 @@ may mark `SessionJudgmentInputs.requests_ordered` to preserve database ordering before public timestamps are converted to whole seconds. The internal `OutcomePrefixPrecondition.verdict_settled_at` accepts fractional epoch seconds; public outcome timestamps keep their existing integer representation. + + +### Inference endpoint and process ownership + +`reflexio.inference_launch` is the import-light shared resolver for provider URLs, +CLI ordinary-child ownership, and enterprise Docker startup/health gating. An +explicit `REFLEXIO_EMBEDDING_SERVICE_URL`, including localhost, attaches to a +caller-owned process. With no URL, the launcher owns the separate inference +child; API workers never load the local model. The enterprise API-local query +child retains its independent reserved route. + +The legacy daemon hostname and disable-local switch are retired. Custom legacy +hostname-only routes or disabled ordinary startup without an endpoint raise a +value-free `ConfigurationMigrationError`; specify the full URL to migrate. +Hostname-based local-service routes used a 30-second request default, while an +explicit URL selects the existing 2-second internal-service default. Set +`REFLEXIO_EMBEDDING_SERVICE_TIMEOUT_MS` explicitly when preserving that older +budget. Shadowed legacy settings do not override a canonical URL. diff --git a/docs/components/configure/config-editor.tsx b/docs/components/configure/config-editor.tsx index 35ee7a325..b8651d789 100644 --- a/docs/components/configure/config-editor.tsx +++ b/docs/components/configure/config-editor.tsx @@ -7,6 +7,7 @@ import { useSettings } from "@/hooks/use-settings"; import { type ReflexioConfig, defaultConfig, + hydrateConfig, serializeConfig, } from "@/lib/config-schema"; import { @@ -29,19 +30,6 @@ type Status = | { kind: "error"; message: string } | { kind: "success"; message: string }; -// Merge server response into our typed defaults. Server may omit optional -// fields; we rely on defaults so subsequent form renders stay stable. -function hydrate(raw: unknown): ReflexioConfig { - const base = defaultConfig(); - if (!raw || typeof raw !== "object") return base; - const incoming = raw as Partial; - return { - ...base, - ...incoming, - agent_success_config: incoming.agent_success_config ?? null, - }; -} - function errorMessage(err: unknown): string { if (err instanceof Error) return err.message; if (typeof err === "string") return err; @@ -69,7 +57,7 @@ export function ConfigEditor() { const body = await res.text(); throw new Error(`GET /api/get_config failed (${res.status}): ${body}`); } - return hydrate(await res.json()); + return hydrateConfig(await res.json()); }, [baseUrl, apiKey]); const load = useCallback(async (signal?: AbortSignal) => { diff --git a/docs/components/configure/sections.tsx b/docs/components/configure/sections.tsx index 21e361e14..ccbee7974 100644 --- a/docs/components/configure/sections.tsx +++ b/docs/components/configure/sections.tsx @@ -19,17 +19,17 @@ import { TextAreaField, } from "./primitives"; import { - APIKeyConfig, - AgentSuccessConfig, + type APIKeyConfig, + type AgentSuccessConfig, EXTRACTION_PRESET_LABELS, - ExtractionPreset, - LLMConfig, + type ExtractionPreset, + type LLMConfig, PRESET_VALUES, - ProfileExtractorConfig, - ReflexioConfig, - StorageConfigSQLite, - ToolUseConfig, - UserPlaybookExtractorConfig, + type ProfileExtractorConfig, + type ReflexioConfig, + type StorageConfigSQLite, + type ToolUseConfig, + type UserPlaybookExtractorConfig, defaultAgentSuccess, defaultPlaybookExtractor, defaultProfileExtractor, @@ -38,6 +38,18 @@ import { type SetConfig = (updater: (prev: ReflexioConfig) => ReflexioConfig) => void; +// Names are editable, so row identity belongs to the UI rather than the payload. +const toolItemIds = new WeakMap(); +let nextToolItemId = 0; +function toolItemId(item: ToolUseConfig): number { + let id = toolItemIds.get(item); + if (id === undefined) { + id = nextToolItemId++; + toolItemIds.set(item, id); + } + return id; +} + export function StorageSection({ value, setConfig, @@ -46,10 +58,20 @@ export function StorageSection({ setConfig: SetConfig; }) { const current = value ?? { db_path: null }; + // Preserve remote/managed snapshots without offering an incompatible SQLite edit. + const editable = Object.entries(current).every( + ([key, field]) => + key === "db_path" || + ((key === "type" || key === "storage_type") && field === "sqlite"), + ); return (
- setConfig((prev) => ({ + editable && setConfig((prev) => ({ ...prev, - storage_config: { db_path: v }, + storage_config: { ...prev.storage_config, db_path: v }, })) } placeholder="(default)" @@ -263,7 +286,7 @@ export function APIKeysSection({ id="key-openai" type="password" value={keys.openai?.api_key ?? null} - onChange={(v) => update({ openai: v ? { api_key: v } : null })} + onChange={(v) => update({ openai: { ...keys.openai, api_key: v } })} placeholder="sk-…" /> @@ -730,7 +753,9 @@ export function ToolsSection({ const updateAt = (idx: number, patch: Partial) => { setConfig((prev) => { const next = [...(prev.tool_can_use ?? [])]; - next[idx] = { ...next[idx], ...patch }; + const previous = next[idx]; + next[idx] = { ...previous, ...patch }; + toolItemIds.set(next[idx], toolItemId(previous)); return { ...prev, tool_can_use: next }; }); }; @@ -761,7 +786,7 @@ export function ToolsSection({ )} {items.map((item, idx) => ( removeAt(idx)} > diff --git a/docs/lib/config-schema.ts b/docs/lib/config-schema.ts index 072354858..932a58025 100644 --- a/docs/lib/config-schema.ts +++ b/docs/lib/config-schema.ts @@ -14,6 +14,12 @@ export interface StorageConfigSQLite { export interface OpenAIConfig { api_key: string | null; + azure_config?: { + api_key: string; + endpoint: string; + api_version: string; + deployment_name?: string | null; + } | null; } export interface SimpleKeyConfig { @@ -194,6 +200,19 @@ export function defaultTool(): ToolUseConfig { return { tool_name: "", tool_description: "" }; } +// Hydration keeps the complete server snapshot while defaults stabilize the UI. +// The editor renders a subset; its full-config save must retain other policies. +export function hydrateConfig(raw: unknown): ReflexioConfig { + const base = defaultConfig(); + if (!raw || typeof raw !== "object") return base; + const incoming = raw as Partial; + return { + ...base, + ...incoming, + agent_success_config: incoming.agent_success_config ?? null, + }; +} + // Strip empty strings → null and empty arrays → null so the server receives // a clean payload. The Pydantic model rejects empty strings for `NonEmptyStr` // fields, so leaving them in would produce confusing 422s. @@ -220,11 +239,14 @@ export function serializeConfig(config: ReflexioConfig): unknown { const entry = out[provider]; if (!entry || clean(entry.api_key) === null) out[provider] = null; } - if (out.openai && clean(out.openai.api_key) === null) out.openai = null; + if (out.openai && clean(out.openai.api_key) === null) { + out.openai = out.openai.azure_config ? { ...out.openai, api_key: null } : null; + } if (out.dashscope && clean(out.dashscope.api_key) === null) { out.dashscope = null; } else if (out.dashscope) { out.dashscope = { + ...out.dashscope, api_key: out.dashscope.api_key, api_base: clean(out.dashscope.api_base), }; @@ -246,6 +268,7 @@ export function serializeConfig(config: ReflexioConfig): unknown { const cleanLlm = (llm: LLMConfig | null): LLMConfig | null => { if (!llm) return null; const out: LLMConfig = { + ...llm, should_run_model_name: clean(llm.should_run_model_name), generation_model_name: clean(llm.generation_model_name), embedding_model_name: clean(llm.embedding_model_name), @@ -298,9 +321,10 @@ export function serializeConfig(config: ReflexioConfig): unknown { }; return { - storage_config: config.storage_config - ? { db_path: clean(config.storage_config.db_path) } - : null, + ...config, + storage_config: config.storage_config && "db_path" in config.storage_config + ? { ...config.storage_config, db_path: clean(config.storage_config.db_path) } + : config.storage_config, agent_context_prompt: clean(config.agent_context_prompt), tool_can_use: config.tool_can_use?.length ? config.tool_can_use : null, profile_extractor_config: cleanExtractor(config.profile_extractor_config), diff --git a/docs/scripts/test-correlation.mjs b/docs/scripts/test-correlation.mjs index 5da227ea4..2cf2154eb 100644 --- a/docs/scripts/test-correlation.mjs +++ b/docs/scripts/test-correlation.mjs @@ -60,6 +60,51 @@ assert.equal(serialized.agent_success_config.retrieved_learning_sampling_rate, 0 assert.equal(serialized.agent_success_config.request_sources_enabled[0], "api"); assert.equal(serialized.profile_extractor_config.window_size_override, 23); +// A subset editor must preserve the fetched snapshot through full replacement. +const fetched = { + ...configSchema.defaultConfig(), + governance_retention: { + audit_events_retention_enabled: false, + audit_events_retention_days: 45, + audit_events_delete_batch_limit: 2000, + }, + lineage_gc: { enabled: false, tombstone_grace_window_days: 120 }, + future_policy: { enabled: false, budget: 37 }, + storage_config: { db_path: " ", future_storage_option: "keep" }, + llm_config: { generation_model_name: "keep-model", future_llm_option: 37 }, + api_key_config: { + openai: { api_key: null, azure_config: { + api_key: "synthetic-azure-key", endpoint: "https://example.openai.azure.com/", + api_version: "2024-02-15-preview", deployment_name: "keep-deployment", + } }, + dashscope: { api_key: "test-key", api_base: "", future_provider_option: 37 }, + }, +}; +const before = JSON.stringify(fetched); +const edited = configSchema.hydrateConfig(fetched); +edited.agent_context_prompt = "An unrelated edit"; +const roundtrip = JSON.parse(JSON.stringify(configSchema.serializeConfig(edited))); +assert.deepEqual(roundtrip.governance_retention, fetched.governance_retention); +assert.deepEqual(roundtrip.lineage_gc, fetched.lineage_gc); +assert.deepEqual(roundtrip.future_policy, fetched.future_policy); +assert.deepEqual(roundtrip.api_key_config.openai, fetched.api_key_config.openai); +assert.equal(roundtrip.agent_context_prompt, "An unrelated edit"); +assert.equal(roundtrip.storage_config.db_path, null); +assert.equal(roundtrip.storage_config.future_storage_option, "keep"); +assert.equal(roundtrip.llm_config.future_llm_option, 37); +assert.equal(roundtrip.api_key_config.dashscope.future_provider_option, 37); +assert.equal(roundtrip.api_key_config.dashscope.api_base, null); +assert.equal(JSON.stringify(fetched), before); +// A snapshot from another storage implementation must not become SQLite. +const remote = { type: "postgres", db_url: "postgresql://local-test", schema: "tenant" }; +assert.deepEqual( + configSchema.serializeConfig(configSchema.hydrateConfig({ ...fetched, storage_config: remote })).storage_config, + remote, +); +// Explicitly clearing a represented optional section remains a real clear. +edited.llm_config = null; +assert.equal(configSchema.serializeConfig(edited).llm_config, null); + const { generatePythonCode } = load("execution/code-generator.ts"); const { initialParams } = load("execution/default-params.ts"); const { unifiedSearchMethods } = load("methods/unified-search.ts"); diff --git a/reflexio/cli/README.md b/reflexio/cli/README.md index 44d35710d..779796563 100644 --- a/reflexio/cli/README.md +++ b/reflexio/cli/README.md @@ -100,8 +100,12 @@ uv run reflexio services stop uv run reflexio services stop --force # SIGKILL instead of SIGTERM ``` +Provider URL resolution and launcher ownership share the pure +`reflexio.inference_launch` resolver. An explicit endpoint, including localhost, +belongs to its existing owner; see `developer.md` for legacy-setting migration. + `services start --only backend` starts the separate `embedding` process unless -`REFLEXIO_EMBEDDING_SERVICE_URL` routes inference to a remote service. This is +`REFLEXIO_EMBEDDING_SERVICE_URL` supplies an explicit endpoint. This is also true when an organization uses cloud embeddings because the service owns reranking. The service exposes `POST /v1/embeddings` and `POST /v1/rerank`, and supports `local/nomic-embed-v1.5`, diff --git a/reflexio/cli/bootstrap_config.py b/reflexio/cli/bootstrap_config.py index 810dafe14..5509f2ee5 100644 --- a/reflexio/cli/bootstrap_config.py +++ b/reflexio/cli/bootstrap_config.py @@ -14,7 +14,7 @@ import typer from reflexio.config_env import resolve_data_db_url, resolve_postgres_pool_defaults -from reflexio.config_migration import ConfigurationMigrationError +from reflexio.config_migration import StoredConfigurationError from .paths import reflexio_home @@ -117,7 +117,7 @@ def load_storage_from_config( storage = LocalFileConfigStorage(org_id, base_dir=base_dir) config = storage.load_config() - except ConfigurationMigrationError: + except StoredConfigurationError: raise except Exception: logger.debug("Failed to load config from %s", config_path, exc_info=True) diff --git a/reflexio/cli/run_services.py b/reflexio/cli/run_services.py index a425a67f4..02895a15d 100644 --- a/reflexio/cli/run_services.py +++ b/reflexio/cli/run_services.py @@ -13,6 +13,7 @@ import reflexio from reflexio.cli.env_loader import load_reflexio_env from reflexio.cli.utils import ServiceConfig, get_env_port, run_services +from reflexio.inference_launch import ordinary_inference_launch logger = logging.getLogger(__name__) @@ -297,13 +298,17 @@ def parse_only_flag(only: str | None, default_services: set[str]) -> set[str]: """ if only: return {s.strip() for s in only.split(",")} - return default_services + return set(default_services) def should_start_local_embedding_service() -> bool: - """Return True unless inference is explicitly routed to a remote service.""" - remote_url = os.environ.get("REFLEXIO_EMBEDDING_SERVICE_URL", "").strip() - return not remote_url + """Return whether the launcher owns the ordinary colocated child.""" + return ( + ordinary_inference_launch( + os.environ, default_port=DEFAULT_OSS_PORTS["embedding"] + ).mode + == "colocated" + ) def _ensure_nextjs_dependencies(project_dir: Path) -> bool: diff --git a/reflexio/config_migration.py b/reflexio/config_migration.py index 7e825d728..b0790881d 100644 --- a/reflexio/config_migration.py +++ b/reflexio/config_migration.py @@ -1,5 +1,9 @@ """Errors that stored-configuration loaders must never turn into defaults.""" -class ConfigurationMigrationError(ValueError): +class StoredConfigurationError(ValueError): + """The saved configuration must be corrected before it can load.""" + + +class ConfigurationMigrationError(StoredConfigurationError): """An explicit migration is required before this saved policy can load.""" diff --git a/reflexio/durable_learning_policy.py b/reflexio/durable_learning_policy.py index 6705f2c3c..d8057f30c 100644 --- a/reflexio/durable_learning_policy.py +++ b/reflexio/durable_learning_policy.py @@ -1,49 +1,66 @@ -"""Code-owned durable learning timing, shared by library and enterprise schedulers. +"""Validated durable learning timing, shared by library and enterprise schedulers. -This module stays independent of server imports so startup can validate legacy -inputs before constructing workers or importing the application stack. +This module stays independent of server imports so startup can validate the +resolved deployment inputs before constructing workers or importing the stack. """ import math import os import threading from collections.abc import Mapping +from dataclasses import dataclass DISCOVERY_POLL_SECONDS = 2.0 CLAIM_LEASE_SECONDS = 300 -def validate_discovery_timing_env(env: Mapping[str, str]) -> None: - """Reject retired custom timing overrides; tolerate blank or old defaults. +# The heartbeat waits lease/3; storage persists epoch + lease as signed int64. +MAX_DISCOVERY_LEASE_SECONDS = min(((1 << 63) - 1) // 2, int(threading.TIMEOUT_MAX) * 3) + + +@dataclass(frozen=True) +class DiscoveryTiming: + poll_seconds: float + lease_seconds: int + + +def resolve_discovery_timing(env: Mapping[str, str] | None = None) -> DiscoveryTiming: + """Resolve independent pickup/load and crash-recovery policies once. - The accepted numbers describe the retired contract, not tunable policy. - Never silently reinterpret a nondefault deployment override as our default. + The supplied mapping is authoritative, including an empty mapping. Defaults + remain 2/300; every consumer uses the same validation and blank handling. """ - legacy = ( - ( - "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS", - env.get("REFLEXIO_DURABLE_LEARNING_POLL_SECONDS", ""), - 2.0, - ), - ( - "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS", - env.get("REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS", ""), - 300.0, - ), - ) - for name, raw, old_default in legacy: - if not raw.strip(): - continue - try: - matches_default = float(raw) == old_default - except ValueError: - matches_default = False - if not matches_default: - raise ValueError( - f"{name} is retired; remove it from your environment. " - "Durable extraction now owns its discovery timing; custom " - "poll intervals and claim durations are no longer supported." - ) + environment = os.environ if env is None else env + poll_raw = environment.get("REFLEXIO_DURABLE_LEARNING_POLL_SECONDS", "").strip() + lease_raw = environment.get("REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS", "").strip() + try: + poll = float(poll_raw) if poll_raw else float(DISCOVERY_POLL_SECONDS) + except ValueError: + poll = 0.0 + if not math.isfinite(poll) or not 0 < poll <= threading.TIMEOUT_MAX: + raise ValueError( + "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS must be finite, positive " + "and no greater than threading.TIMEOUT_MAX." + ) + try: + lease = float(lease_raw) if lease_raw else float(CLAIM_LEASE_SECONDS) + except ValueError: + lease = 0.0 + if ( + not math.isfinite(lease) + or not 1 <= lease <= MAX_DISCOVERY_LEASE_SECONDS + or not lease.is_integer() + ): + raise ValueError( + "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS must be a positive integer " + "within the signed-64-bit epoch and heartbeat wait range." + ) + return DiscoveryTiming(poll_seconds=poll, lease_seconds=int(lease)) + + +def validate_discovery_timing_env(env: Mapping[str, str]) -> None: + """Validate the resolved deployment mapping before constructing workers.""" + resolve_discovery_timing(env) AGGREGATION_MIN_INTERVAL_SECONDS = 3600 diff --git a/reflexio/env_utils.py b/reflexio/env_utils.py new file mode 100644 index 000000000..4dec7201d --- /dev/null +++ b/reflexio/env_utils.py @@ -0,0 +1,11 @@ +"""Environment parsing shared by import-light launchers and the server.""" + + +def env_truthy(value: str) -> bool: + """Parse the historical permissive boolean format. + + New boolean settings should use ``server.env_utils.env_bool`` for strict + validation. This parser retains compatibility for existing callers and + migration checks without importing the server's startup side effects. + """ + return value.strip().lower() in {"1", "true", "yes", "on"} diff --git a/reflexio/housekeeping_policy.py b/reflexio/housekeeping_policy.py index 8cad4e785..754842e95 100644 --- a/reflexio/housekeeping_policy.py +++ b/reflexio/housekeeping_policy.py @@ -1,4 +1,4 @@ -"""Code-owned housekeeping execution and safe legacy configuration preflight. +"""Housekeeping cadence defaults and safe legacy configuration preflight. No server imports: configuration parsing and image bootstrap use this module. """ @@ -8,13 +8,9 @@ from reflexio.config_migration import ConfigurationMigrationError DAILY_SWEEP_SECONDS = 86_400 -AUDIT_DELETE_BATCH_LIMIT = 500 # Frozen compatibility defaults, independent of future policy calibration. -_RETIRED_FIELDS = ( - ("lineage_gc", "poll_interval_seconds", 86_400), - ("governance_retention", "audit_events_delete_batch_limit", 500), -) +_RETIRED_FIELDS = (("lineage_gc", "poll_interval_seconds", 86_400),) class HousekeepingMigrationError(ConfigurationMigrationError): @@ -44,8 +40,7 @@ def strip_legacy_housekeeping_fields(payload: dict[str, Any]) -> dict[str, Any]: raise HousekeepingMigrationError( "Retired housekeeping settings require explicit migration: " + ", ".join(conflicts) - + ". Remove these fields to adopt code-owned daily sweeps and " - "500-row deletion batches per table." + + ". Remove these fields to adopt code-owned daily sweeps." ) cleaned = dict(payload) for section, field, _ in _RETIRED_FIELDS: diff --git a/reflexio/inference_launch.py b/reflexio/inference_launch.py new file mode 100644 index 000000000..e0f235d42 --- /dev/null +++ b/reflexio/inference_launch.py @@ -0,0 +1,83 @@ +"""Pure inference endpoint and launcher ownership; no model or server imports.""" + +from __future__ import annotations + +import os +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Literal + +from reflexio.config_migration import ConfigurationMigrationError +from reflexio.env_utils import env_truthy + +_SERVICE_URL = "REFLEXIO_EMBEDDING_SERVICE_URL" +_LEGACY_HOST = "REFLEXIO_EMBEDDING_DAEMON_HOST" +_LEGACY_DISABLE = "REFLEXIO_DISABLE_LOCAL_EMBEDDING_DAEMON" +_QUERY_URL = "REFLEXIO_QUERY_EMBEDDING_SERVICE_URL" +_API_QUERY_URL = "http://127.0.0.1:8093" + + +@dataclass(frozen=True) +class InferenceLaunchPlan: + """The owned process role and its startup health gate, if any.""" + + mode: Literal["colocated", "attached", "query"] + health_url: str | None + + +def inference_service_url(env: Mapping[str, str], *, default_port: int = 8072) -> str: + """Resolve one endpoint; custom legacy hosts require explicit migration.""" + explicit = env.get(_SERVICE_URL, "").strip() + if explicit: + return explicit.rstrip("/") + if env.get(_LEGACY_HOST, "").strip(): + raise ConfigurationMigrationError( + f"{_LEGACY_HOST} is retired. Configure {_SERVICE_URL} with the full " + "endpoint URL and choose REFLEXIO_EMBEDDING_SERVICE_TIMEOUT_MS " + "explicitly if retaining the previous local-service request budget." + ) + port = env.get("EMBEDDING_PORT", "").strip() or str(default_port) + return f"http://127.0.0.1:{port}" + + +def ordinary_inference_launch( + env: Mapping[str, str], *, default_port: int = 8072 +) -> InferenceLaunchPlan: + """An explicit endpoint has its own owner; otherwise own the local child.""" + endpoint = inference_service_url(env, default_port=default_port) + if env.get(_SERVICE_URL, "").strip(): + return InferenceLaunchPlan("attached", None) + if env_truthy(env.get(_LEGACY_DISABLE, "")): + raise ConfigurationMigrationError( + f"{_LEGACY_DISABLE} is retired. To attach to an existing process, " + f"configure {_SERVICE_URL} explicitly, including for localhost. " + "Choose REFLEXIO_EMBEDDING_SERVICE_TIMEOUT_MS explicitly if " + "retaining the previous local-service request budget." + ) + return InferenceLaunchPlan("colocated", endpoint + "/health") + + +def container_inference_launch(env: Mapping[str, str]) -> InferenceLaunchPlan: + """Resolve Docker's independent query role before ordinary child ownership.""" + if env.get("REFLEXIO_API_QUERY_EMBEDDING_ENABLED", "false") == "true": + if env.get(_QUERY_URL, "").strip().rstrip("/") != _API_QUERY_URL: + raise ConfigurationMigrationError( + "API-local query mode requires its reserved loopback URL." + ) + # The ordinary disable switch never selected this independent child. + inference_service_url(env, default_port=8089) + return InferenceLaunchPlan("query", _API_QUERY_URL + "/health/model") + return ordinary_inference_launch(env, default_port=8089) + + +def main() -> None: + """Small shell adapter: fixed mode and health URL, never eval shell code.""" + try: + plan = container_inference_launch(os.environ) + except ConfigurationMigrationError as exc: + raise SystemExit(str(exc)) from None + print(plan.mode, plan.health_url or "-") + + +if __name__ == "__main__": + main() diff --git a/reflexio/models/config_schema.py b/reflexio/models/config_schema.py index 46fe42b9d..091eb2676 100644 --- a/reflexio/models/config_schema.py +++ b/reflexio/models/config_schema.py @@ -17,7 +17,10 @@ DEFAULT_STORAGE_POOL_ACQUIRE_TIMEOUT, DEFAULT_STORAGE_POOL_SIZE, ) -from reflexio.config_migration import ConfigurationMigrationError +from reflexio.config_migration import ( + ConfigurationMigrationError, + StoredConfigurationError, +) from reflexio.housekeeping_policy import strip_legacy_housekeeping_fields from reflexio.pending_tool_policy import strip_legacy_pending_tool_timing @@ -127,6 +130,7 @@ class ExtractionPreset(StrEnum): ExtractionPreset.LONG_FORM: (25, 10), ExtractionPreset.HIGH_VOLUME: (15, 8), } +_LEGACY_EXTRACTION_PRESET = TypeAdapter(ExtractionPreset | None) # --------------------------------------------------------------------------- @@ -230,6 +234,19 @@ def strip_retired_config_fields(data: dict[str, Any]) -> dict[str, Any]: can share this cleanup without mutating a caller's saved configuration. """ payload = strip_legacy_housekeeping_fields(data) + if "extraction_preset" in payload: + # A preset was an input shortcut, never independent execution policy. + # Resolve it before raw upgrades can preserve it as opaque metadata. + payload = _migrate_dict(payload, _CONFIG_FIELD_MIGRATION) + preset = _LEGACY_EXTRACTION_PRESET.validate_python( + payload.pop("extraction_preset") + ) + if preset is not None: + for name, value in zip( + ("window_size", "stride_size"), _PRESET_VALUES[preset], strict=True + ): + if payload.get(name) is None: + payload[name] = value if "pending_tool_call_config" in payload: payload["pending_tool_call_config"] = strip_legacy_pending_tool_timing( payload["pending_tool_call_config"] @@ -373,8 +390,14 @@ class StorageConfigSupabase(BaseModel): key: NonEmptyStr db_url: NonEmptyStr schema_name: str | None = Field(default=None, alias="schema") - read_url: NonEmptyStr | None = None - read_key: NonEmptyStr | None = None + + @model_validator(mode="before") + @classmethod + def retire_http_reader(cls, data: Any) -> Any: + """Old HTTP reader fields no longer affect native SQL read execution.""" + if isinstance(data, dict): + return {k: v for k, v in data.items() if k not in {"read_url", "read_key"}} + return data class StorageConfigPostgres(BaseModel): @@ -909,13 +932,12 @@ class GovernanceRetentionConfig(BaseModel): audit_events_retention_enabled: bool = False audit_events_retention_days: int = Field(default=365, gt=0) - @model_validator(mode="before") - @classmethod - def discard_legacy_execution_settings(cls, data: Any) -> Any: - """Accept former defaults, reject custom execution policy inputs.""" - return strip_legacy_housekeeping_fields({"governance_retention": data})[ - "governance_retention" - ] + audit_events_delete_batch_limit: int = Field( + default=500, + gt=0, + le=(1 << 31) - 1, + description="Maximum expired rows deleted per table in one governance sweep.", + ) @dataclass(frozen=True) @@ -1069,8 +1091,6 @@ class Config(BaseModel): agent_success_config: AgentSuccessConfig | None = Field( default_factory=_default_agent_success_config ) - # extraction preset — selects bundled window_size/stride_size values - extraction_preset: ExtractionPreset | None = None # extraction parameters window_size: int = Field(default=DEFAULT_WINDOW_SIZE, gt=0) stride_size: int = Field(default=DEFAULT_STRIDE_SIZE, gt=0) @@ -1167,28 +1187,6 @@ def _migrate_field_names(cls, data: Any) -> Any: del data[key] return data - @model_validator(mode="after") - def apply_extraction_preset(self) -> Self: - """Apply preset values when window_size/stride_size are at defaults. - - If a preset is selected but the user also explicitly set window_size or - stride_size, the explicit values win (checked via model_fields_set). - """ - if self.extraction_preset is None: - return self - - preset_values = _PRESET_VALUES.get(self.extraction_preset) - if preset_values is None: - return self - - preset_window_size, preset_stride_size = preset_values - if "window_size" not in self.model_fields_set: - self.window_size = preset_window_size - if "stride_size" not in self.model_fields_set: - self.stride_size = preset_stride_size - - return self - @model_validator(mode="after") def check_stride_size_le_window_size(self) -> Self: """Validate that stride_size <= window_size.""" @@ -1250,23 +1248,42 @@ def validate_stored_config(data: dict[str, Any]) -> Config: Persisted JSON can contain fields that were valid when it was written but have since been deleted. Ignore only those unknown fields while retaining normal validation for recognized values. Missing fields continue to use - their current schema defaults. + their current schema defaults. Invalid recognized values raise a sanitized + StoredConfigurationError so loaders cannot replace saved policy with defaults. API writes intentionally construct ``Config`` directly and therefore keep the model's strict ``extra="forbid"`` behavior. """ - # Run retirement preflight outside Pydantic so its typed migration error - # survives; otherwise ValueError is wrapped and mistaken for recoverable - # schema drift by storage loaders. - normalized = strip_retired_config_fields(normalize_legacy_config_shape(data)) - # Validate storage_config on its own, strictly, BEFORE the lenient pass - # below. ``extra="ignore"`` cascades into nested models, so it overrides each - # StorageConfig member's ``extra="forbid"`` -- without this pre-check a - # Supabase config that lost its ``db_url`` validates as SQLite here and the - # org is silently rerouted to a local file. Schema-evolution leniency is - # meant for Config's own retired fields, not for reinterpreting which - # storage backend an org is on. - storage_config = normalized.get("storage_config") - if isinstance(storage_config, dict): - _STORAGE_CONFIG_ADAPTER.validate_python(storage_config) - return Config.model_validate(normalized, extra="ignore") + try: + # Normalize before Pydantic model construction so typed migration errors + # remain intact. Pydantic failures from legacy conversion need the same + # fail-closed boundary as failures in the resulting current schema. + normalized = strip_retired_config_fields(normalize_legacy_config_shape(data)) + # extra="ignore" cascades into nested models; validate storage strictly + # first so a malformed remote config cannot become a local backend. + storage_config = normalized.get("storage_config") + if isinstance(storage_config, dict): + try: + _STORAGE_CONFIG_ADAPTER.validate_python(storage_config) + except ValidationError: + raise StoredConfigurationError( + "Stored configuration failed validation at storage_config. " + "Correct the saved values before continuing." + ) from None + return Config.model_validate(normalized, extra="ignore") + except ValidationError as exc: + # Only the first location component is an authoritative Config field. + # Nested components can be supplied map keys, even when their spelling + # happens to match a property declared elsewhere in the schema. + paths = sorted( + { + str(error["loc"][0]) + for error in exc.errors(include_input=False, include_context=False) + if error["loc"] and error["loc"][0] in Config.model_fields + } + ) or ["configuration"] + raise StoredConfigurationError( + "Stored configuration failed validation at " + + ", ".join(paths[:10]) + + ". Correct the saved values before continuing." + ) from None diff --git a/reflexio/server/env_utils.py b/reflexio/server/env_utils.py index 2dc585d58..057184999 100644 --- a/reflexio/server/env_utils.py +++ b/reflexio/server/env_utils.py @@ -6,6 +6,8 @@ import os from collections.abc import Mapping +from reflexio.env_utils import env_truthy as env_truthy + def env_get(env: Mapping[str, str], name: str) -> str: """Return a stripped environment value or an empty string.""" @@ -126,17 +128,6 @@ def env_bool( ) -def env_truthy(value: str) -> bool: - """Return whether a string environment value is truthy. - - Deprecated in favour of :func:`env_bool`, which takes the variable NAME and - can therefore name it when refusing a bad value. Retained because it is a - public export of this package; it keeps the historical permissive set so an - external caller's behaviour does not change silently. - """ - return value.strip().lower() in {"1", "true", "yes", "on"} - - def positive_int_env(name: str, default: int, logger: logging.Logger) -> int: """Resolve a strictly-positive int from environment variable ``name``. diff --git a/reflexio/server/llm/providers/embedding_service_provider.py b/reflexio/server/llm/providers/embedding_service_provider.py index 010d09706..9d243877e 100644 --- a/reflexio/server/llm/providers/embedding_service_provider.py +++ b/reflexio/server/llm/providers/embedding_service_provider.py @@ -21,6 +21,9 @@ import httpx +from reflexio.inference_launch import ( + inference_service_url as resolve_inference_service_url, +) from reflexio.server.env_utils import env_str, env_truthy, positive_int_env from reflexio.server.search_runtime import ( checkpoint, @@ -41,12 +44,9 @@ _query_contract_cache: dict[tuple[int, str, str], tuple[str, str, int]] = {} _query_contract_lock = threading.Lock() _ENV_TIMEOUT_MS = "REFLEXIO_EMBEDDING_SERVICE_TIMEOUT_MS" -_ENV_EMBEDDING_PORT = "EMBEDDING_PORT" -_ENV_DAEMON_HOST = "REFLEXIO_EMBEDDING_DAEMON_HOST" _ENV_CLAUDE_SMART_LOCAL = "CLAUDE_SMART_USE_LOCAL_EMBEDDING" _ENV_MAX_TEXTS_PER_REQUEST = "REFLEXIO_EMBEDDING_SERVICE_MAX_TEXTS_PER_REQUEST" CUSTOM_EMBEDDING_MODEL = "custom" -_DEFAULT_LOCAL_PORT = 8072 _DEFAULT_INTERNAL_SERVICE_TIMEOUT_MS = 2_000 _DEFAULT_LOCAL_SERVICE_TIMEOUT_MS = 30_000 # Each request embeds its whole ``input`` list in one ``model.encode()`` on the @@ -83,12 +83,6 @@ class InferenceServiceCapabilities: workload: str = "shared" -def _local_service_url() -> str: - host = os.environ.get(_ENV_DAEMON_HOST, "").strip() or "127.0.0.1" - port = os.environ.get(_ENV_EMBEDDING_PORT, str(_DEFAULT_LOCAL_PORT)) - return f"http://{host}:{port}" - - def inference_service_url() -> str: """Return the shared local/internal inference-service base URL. @@ -96,8 +90,7 @@ def inference_service_url() -> str: reranking can share the same daemon even when an organization uses cloud embeddings. """ - configured = os.environ.get(_ENV_SERVICE_URL, "").strip() - return configured.rstrip("/") if configured else _local_service_url() + return resolve_inference_service_url(os.environ) def remote_inference_service_configured() -> bool: @@ -395,9 +388,6 @@ def embedding_provider_mode(model: str | None = None) -> EmbeddingProviderMode: if os.environ.get(_ENV_SERVICE_URL, "").strip(): return "internal_service" - if _uses_embedding_service(model) and os.environ.get(_ENV_DAEMON_HOST, "").strip(): - return "local_service" - if _uses_embedding_service(model): return "local_service" return "cloud" diff --git a/reflexio/server/services/configurator/base_configurator.py b/reflexio/server/services/configurator/base_configurator.py index ec26f3372..886c824f2 100644 --- a/reflexio/server/services/configurator/base_configurator.py +++ b/reflexio/server/services/configurator/base_configurator.py @@ -135,6 +135,11 @@ def set_config_by_name( config_name: str, config_value: str | int | float | bool | list | dict | BaseModel | None, ) -> None: + if config_name == "extraction_preset": + # Old named writes only changed the tag: existing explicit geometry + # still won on reload. Keep that precedence without persisting a tag. + self.set_config(self.prepare_config_patch({config_name: config_value})) + return original_config_name = config_name config_name = _CONFIG_NAME_ALIASES.get(config_name, config_name) if config_name not in type(self.config).model_fields: diff --git a/reflexio/server/services/configurator/local_file_config_storage.py b/reflexio/server/services/configurator/local_file_config_storage.py index fc28de8ea..a05d849e4 100644 --- a/reflexio/server/services/configurator/local_file_config_storage.py +++ b/reflexio/server/services/configurator/local_file_config_storage.py @@ -8,7 +8,7 @@ from typing import Any from reflexio.cli.paths import reflexio_home -from reflexio.config_migration import ConfigurationMigrationError +from reflexio.config_migration import StoredConfigurationError from reflexio.models.config_schema import ( Config, PlaybookConfig, @@ -116,7 +116,8 @@ def load_config(self) -> Config: self.config_file, ) return config - except ConfigurationMigrationError: + except StoredConfigurationError as exc: + logger.error("Invalid stored config at %s: %s", self.config_file, exc) raise except Exception: logger.exception( diff --git a/reflexio/server/services/durable_learning/scheduler.py b/reflexio/server/services/durable_learning/scheduler.py index d1d49841c..5093d0e79 100644 --- a/reflexio/server/services/durable_learning/scheduler.py +++ b/reflexio/server/services/durable_learning/scheduler.py @@ -3,7 +3,6 @@ from __future__ import annotations import logging -import os from collections.abc import Callable, Iterable from reflexio import durable_learning_policy @@ -29,14 +28,14 @@ def __init__( org_ids_provider: Callable[[], Iterable[str]], instance_id: str | None = None, ): - durable_learning_policy.validate_discovery_timing_env(os.environ) + timing = durable_learning_policy.resolve_discovery_timing() super().__init__(thread_name="reflexio-durable-learning-scheduler") self._worker = DurableLearningWorker( request_context_factory, instance_id=instance_id ) self._org_ids_provider = org_ids_provider - self._poll = durable_learning_policy.DISCOVERY_POLL_SECONDS - self._lease = durable_learning_policy.CLAIM_LEASE_SECONDS + self._poll = timing.poll_seconds + self._lease = timing.lease_seconds self._last_org: str | None = None def _run_once(self) -> float: diff --git a/tests/conftest.py b/tests/conftest.py index 3900e56b8..7100c7dff 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -233,8 +233,8 @@ def pytest_unconfigure(config): @pytest.fixture(autouse=True) def _fast_discovery_policy(monkeypatch: pytest.MonkeyPatch) -> None: - """Speed internal scheduler ticks without reviving a retired public knob.""" - monkeypatch.setattr("reflexio.durable_learning_policy.DISCOVERY_POLL_SECONDS", 0.05) + """Exercise the supported process policy while keeping discovery tests fast.""" + monkeypatch.setenv("REFLEXIO_DURABLE_LEARNING_POLL_SECONDS", "0.05") @pytest.fixture(autouse=True) diff --git a/tests/models/test_extraction_geometry.py b/tests/models/test_extraction_geometry.py new file mode 100644 index 000000000..bc348f4ef --- /dev/null +++ b/tests/models/test_extraction_geometry.py @@ -0,0 +1,118 @@ +"""Window/stride are the sole persisted extraction geometry authority.""" + +import json +from copy import deepcopy +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from reflexio.models.config_schema import ( + Config, + strip_retired_config_fields, + validate_stored_config, +) +from reflexio.server.services.configurator.configurator import DefaultConfigurator +from reflexio.server.services.configurator.local_file_config_storage import ( + LocalFileConfigStorage, +) + + +@pytest.mark.parametrize( + "preset,geometry", + [ + ("quick_chat", (5, 3)), + ("standard", (10, 8)), + ("long_form", (25, 10)), + ("high_volume", (15, 8)), + (None, (10, 8)), + ], +) +@pytest.mark.parametrize("loader", [Config.model_validate, validate_stored_config]) +def test_legacy_shortcuts_resolve_once_without_persisted_authority( + preset, geometry, loader +): + payload = {"storage_config": None, "extraction_preset": preset} + before = deepcopy(payload) + config = loader(payload) + assert (config.window_size, config.stride_size) == geometry + assert payload == before + dumped = config.model_dump(mode="json") + assert "extraction_preset" not in dumped + assert "extraction_preset" not in Config.model_json_schema()["properties"] + assert Config.model_validate(dumped) == config + + +@pytest.mark.parametrize( + "fields,geometry", + [ + ({"window_size": 42}, (42, 10)), + ({"stride_size": 3}, (25, 3)), + ({"window_size": 42, "stride_size": 3}, (42, 3)), + ({"window_size": None, "stride_size": None}, (25, 10)), + ({"extraction_window_size": 42, "batch_interval": 3}, (42, 3)), + ], +) +def test_explicit_geometry_and_old_field_aliases_keep_precedence(fields, geometry): + config = Config.model_validate( + {"storage_config": None, "extraction_preset": "long_form", **fields} + ) + assert (config.window_size, config.stride_size) == geometry + + +@pytest.mark.parametrize("preset", ["unknown", 42, {}, False]) +def test_invalid_legacy_shortcuts_are_not_silently_dropped(preset): + with pytest.raises(ValidationError): + Config.model_validate( + { + "storage_config": None, + "extraction_preset": preset, + "window_size": 42, + "stride_size": 3, + } + ) + + +def test_raw_cleanup_preserves_extensions_and_removes_preset_before_upgrade(): + payload = { + "storage_config": None, + "extraction_preset": "long_form", + "extension": {"opaque": [1, 2]}, + } + before = deepcopy(payload) + cleaned = strip_retired_config_fields(payload) + assert payload == before + assert cleaned == { + "storage_config": None, + "window_size": 25, + "stride_size": 10, + "extension": {"opaque": [1, 2]}, + } + assert strip_retired_config_fields(cleaned) == cleaned + + +def test_real_config_file_load_patch_save_and_reload(tmp_path: Path): + store = LocalFileConfigStorage("geometry", base_dir=str(tmp_path)) + path = Path(store.config_file) + path.parent.mkdir(parents=True) + path.write_text( + json.dumps({"storage_config": None, "extraction_preset": "long_form"}) + ) + configurator = DefaultConfigurator("geometry", base_dir=str(tmp_path)) + assert configurator.get_config().window_size == 25 + # Shallow patches already contain explicit geometry: an old label alone + # must not gain new authority over it. + updated = configurator.prepare_config_patch({"extraction_preset": "quick_chat"}) + assert (updated.window_size, updated.stride_size) == (25, 10) + configurator.set_config(updated) + configurator.set_config_by_name("extraction_preset", "standard") + assert "extraction_preset" not in json.loads(path.read_text()) + reloaded = store.load_config() + assert reloaded is not None + assert (reloaded.window_size, reloaded.stride_size) == (25, 10) + configurator.set_config( + configurator.prepare_config_patch({"window_size": 31, "stride_size": 7}) + ) + reloaded = store.load_config() + assert reloaded is not None + assert (reloaded.window_size, reloaded.stride_size) == (31, 7) diff --git a/tests/models/test_housekeeping_policy.py b/tests/models/test_housekeeping_policy.py index d654a020d..4a51cc0e3 100644 --- a/tests/models/test_housekeeping_policy.py +++ b/tests/models/test_housekeeping_policy.py @@ -20,12 +20,6 @@ "section,field,default,model", [ ("lineage_gc", "poll_interval_seconds", 86400, LineageGCConfig), - ( - "governance_retention", - "audit_events_delete_batch_limit", - 500, - GovernanceRetentionConfig, - ), ], ) def test_legacy_defaults_are_accepted_but_not_exposed(section, field, default, model): @@ -41,7 +35,6 @@ def test_legacy_defaults_are_accepted_but_not_exposed(section, field, default, m "section,field", [ ("lineage_gc", "poll_interval_seconds"), - ("governance_retention", "audit_events_delete_batch_limit"), ], ) @pytest.mark.parametrize( @@ -65,7 +58,7 @@ def test_cleanup_preserves_policy_holds_and_opaque_metadata_without_mutating(): "governance_retention": { "audit_events_retention_enabled": False, "audit_events_retention_days": 30, - "audit_events_delete_batch_limit": 500, + "audit_events_delete_batch_limit": 2000, }, "opaque": {"poll_interval_seconds": 60, "audit_events_delete_batch_limit": 1}, } @@ -80,16 +73,13 @@ def test_cleanup_preserves_policy_holds_and_opaque_metadata_without_mutating(): assert cfg.lineage_gc.tombstone_grace_window_days == 120 assert cfg.governance_retention.audit_events_retention_enabled is False assert cfg.governance_retention.audit_events_retention_days == 30 + assert cfg.governance_retention.audit_events_delete_batch_limit == 2000 @pytest.mark.parametrize( "model,payload", [ (LineageGCConfig, {"poll_interval_seconds": "private-legacy-value"}), - ( - GovernanceRetentionConfig, - {"audit_events_delete_batch_limit": "private-legacy-value"}, - ), ( Config, { @@ -106,3 +96,62 @@ def test_migration_error_text_does_not_echo_config_values(model, payload): assert "explicit migration" in str(exc.value) assert "private-legacy-value" not in str(exc.value) assert "private-context" not in str(exc.value) + + +@pytest.mark.parametrize("limit", [1, 2, 500, 2000, (1 << 31) - 1]) +def test_supported_audit_batch_survives_cleanup_and_roundtrip(limit): + from reflexio.models.config_schema import validate_stored_config + + payload = { + "storage_config": None, + "lineage_gc": {"enabled": False}, + "governance_retention": { + "audit_events_retention_enabled": False, + "audit_events_retention_days": 45, + "audit_events_delete_batch_limit": limit, + }, + } + assert housekeeping_config_conflicts(payload) == [] + assert strip_legacy_housekeeping_fields(payload) == payload + cfg = validate_stored_config(payload) + assert cfg.governance_retention.audit_events_delete_batch_limit == limit + assert cfg.governance_retention.audit_events_retention_enabled is False + assert cfg.lineage_gc.enabled is False + assert ( + validate_stored_config( + cfg.model_dump(mode="json") + ).governance_retention.audit_events_delete_batch_limit + == limit + ) + + +@pytest.mark.parametrize( + "limit", + [ + 0, + -1, + 1 << 31, + 2.5, + "private-invalid-limit", + None, + float("inf"), + float("nan"), + {}, + [], + ], +) +def test_invalid_supported_batch_fails_without_disclosing_values(limit): + from reflexio.config_migration import StoredConfigurationError + from reflexio.models.config_schema import validate_stored_config + + payload = { + "storage_config": None, + "governance_retention": {"audit_events_delete_batch_limit": limit}, + } + with pytest.raises(StoredConfigurationError, match="governance_retention") as error: + validate_stored_config(payload) + assert "private-invalid-limit" not in str(error.value) + + +def test_audit_batch_defaults_to_previous_500_rows(): + assert GovernanceRetentionConfig().audit_events_delete_batch_limit == 500 diff --git a/tests/models/test_storage_config_union.py b/tests/models/test_storage_config_union.py index 156137431..0e9f2eb70 100644 --- a/tests/models/test_storage_config_union.py +++ b/tests/models/test_storage_config_union.py @@ -23,6 +23,7 @@ import pytest from pydantic import TypeAdapter, ValidationError +from reflexio.config_migration import StoredConfigurationError from reflexio.models.config_schema import ( StorageConfig, StorageConfigPostgres, @@ -162,7 +163,7 @@ class TestThroughValidateStoredConfig: def test_malformed_storage_config_is_rejected_on_the_load_path( self, label: str, storage_config: dict[str, object] ) -> None: - with pytest.raises(ValidationError): + with pytest.raises(StoredConfigurationError): validate_stored_config({"storage_config": storage_config}) def test_valid_storage_configs_survive_the_load_path(self) -> None: @@ -246,7 +247,9 @@ def test_serialized_shape_carries_no_type_tag( f"{label}: serializing the tag would rewrite every org's stored " f"config blob at the next boot upgrade" ) - assert set(dumped) == set(payload) | {k for k in dumped if dumped[k] is None} + assert set(dumped) == (set(payload) - {"read_url", "read_key"}) | { + k for k in dumped if dumped[k] is None + } def test_round_trip_preserves_the_backend() -> None: @@ -267,3 +270,69 @@ def test_round_trip_preserves_the_backend() -> None: f"round trip changed backend: {type(original).__name__} -> " f"{type(reloaded).__name__}" ) + + +@pytest.mark.parametrize( + "private_key", + ["private-tool-map-key", "enabled", "model", "db_url", "similarity_threshold"], +) +def test_stored_policy_error_and_traceback_omit_values_and_dynamic_keys(private_key): + import traceback + + secret = "private-customer-content" + payload = { + "storage_config": None, + "agent_context_prompt": secret, + "pending_tool_call_config": { + "tool_overrides": {private_key: {"similarity_threshold": 2}}, + }, + } + try: + validate_stored_config(payload) + except StoredConfigurationError as exc: + rendered = "".join(traceback.format_exception(exc)) + assert secret not in rendered + assert f".{private_key}." not in str(exc) + assert "tool_overrides" not in str(exc) + assert "pending_tool_call_config" in str(exc) + if private_key == "private-tool-map-key": + assert private_key not in rendered + assert "input_value" not in rendered + else: + pytest.fail("invalid recognized saved policy was accepted") + + +@pytest.mark.parametrize("value", [-1, "private-invalid-merge-budget"]) +def test_invalid_current_value_in_legacy_conversion_also_fails_closed(value): + import traceback + + payload = { + "storage_config": None, + "playbook_optimizer_config": { + "use_merge": True, + "max_merge_invocations": value, + }, + } + with pytest.raises(StoredConfigurationError) as error: + validate_stored_config(payload) + assert "private-invalid-merge-budget" not in "".join( + traceback.format_exception(error.value) + ) + + +@pytest.mark.parametrize( + "read_url,read_key", [(None, None), ("https://unused.example", "unused"), ("", "")] +) +def test_obsolete_http_reader_drops_without_changing_storage_identity( + read_url, read_key +): + payload = {**_PROD_SUPABASE, "read_url": read_url, "read_key": read_key} + before = dict(payload) + config = validate_stored_config({"storage_config": payload}) + assert isinstance(config.storage_config, StorageConfigSupabase) + canonical = config.storage_config.model_dump(mode="json") + assert canonical == {k: v for k, v in payload.items() if not k.startswith("read_")} + assert payload == before + assert _ADAPTER.validate_python(canonical) == config.storage_config + assert "read_url" not in StorageConfigSupabase.model_fields + assert "read_key" not in StorageConfigSupabase.model_fields diff --git a/tests/server/services/configurator/test_config_storage_contract.py b/tests/server/services/configurator/test_config_storage_contract.py index fb37b6312..9fa0a2028 100644 --- a/tests/server/services/configurator/test_config_storage_contract.py +++ b/tests/server/services/configurator/test_config_storage_contract.py @@ -261,7 +261,6 @@ def test_legacy_shadow_file_load_patch_save_roundtrip(tmp_path: Path) -> None: ("playbook_optimizer_config", "abort_cooldown_threshold"), ("playbook_optimizer_config", "cooldown_after_aborts_seconds"), ("lineage_gc", "poll_interval_seconds"), - ("governance_retention", "audit_events_delete_batch_limit"), ], ) def test_retired_custom_policy_propagates_without_resetting_file( @@ -293,3 +292,52 @@ def test_retired_custom_policy_propagates_without_resetting_file( load() assert "private-custom-policy" not in str(error.value) assert path.read_bytes() == before + + +@pytest.mark.parametrize( + "invalid_policy", + [ + {"audit_events_retention_days": 0}, + {"audit_events_delete_batch_limit": 0}, + ], +) +def test_invalid_recognized_policy_preserves_file_and_blocks_cli_fallback( + tmp_path, invalid_policy +): + from reflexio.cli.bootstrap_config import ( + load_storage_from_config, + save_storage_to_config, + ) + from reflexio.config_migration import StoredConfigurationError + + storage = _make_local_file_storage(str(tmp_path)) + payload = { + "storage_config": {"type": "sqlite"}, + "lineage_gc": {"enabled": False}, + "agent_context_prompt": "private-customer-prompt", + "governance_retention": invalid_policy, + } + path = Path(storage.config_file) + path.parent.mkdir(parents=True) + path.write_text(json.dumps(payload)) + original = path.read_bytes() + for load in ( + storage.load_config, + lambda: load_storage_from_config("test-org", base_dir=str(tmp_path)), + lambda: save_storage_to_config("sqlite", "test-org", base_dir=str(tmp_path)), + ): + with pytest.raises(StoredConfigurationError, match="governance_retention"): + load() + assert path.read_bytes() == original + payload["governance_retention"] = { + "audit_events_retention_days": 45, + "audit_events_delete_batch_limit": 2, + } + path.write_text(json.dumps(payload)) + cfg = storage.load_config() + assert cfg.lineage_gc.enabled is False + assert cfg.governance_retention.audit_events_retention_days == 45 + storage.save_config(cfg) + reloaded = storage.load_config() + assert reloaded.lineage_gc.enabled is False + assert reloaded.governance_retention.audit_events_delete_batch_limit == 2 diff --git a/tests/server/services/durable_learning/test_scheduler.py b/tests/server/services/durable_learning/test_scheduler.py index 1d3a48eee..5454c7609 100644 --- a/tests/server/services/durable_learning/test_scheduler.py +++ b/tests/server/services/durable_learning/test_scheduler.py @@ -2,6 +2,8 @@ import time +import pytest + from reflexio.server.api_endpoints.request_context import RequestContext from reflexio.server.services.durable_learning.scheduler import DurableLearningScheduler @@ -85,7 +87,17 @@ def test_library_recovers_persisted_backlog_without_new_publish(tmp_path, monkey monkeypatch.setattr( ProfileGenerationService, "_should_run_before_extraction", lambda *_: False ) - monkeypatch.setattr("reflexio.durable_learning_policy.DISCOVERY_POLL_SECONDS", 0.05) + monkeypatch.setenv("REFLEXIO_DURABLE_LEARNING_POLL_SECONDS", "0.05") + monkeypatch.setenv("REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS", "45") + storage_type = type(first.get_storage()) + claim = storage_type.claim_extraction + leases = [] + + def observe_claim(storage, owner, lease_seconds): + leases.append(lease_seconds) + return claim(storage, owner, lease_seconds) + + monkeypatch.setattr(storage_type, "claim_extraction", observe_claim) # A new facade opens the existing DB and discovers its work at construction. second = Reflexio(org_id=org, storage_base_dir=str(tmp_path), configurator=config) try: @@ -96,6 +108,7 @@ def test_library_recovers_persisted_backlog_without_new_publish(tmp_path, monkey ): time.sleep(0.02) assert second.get_storage().extraction_status("u", "r")["status"] == "done" + assert leases and set(leases) == {45} finally: with local._lock: scheduler = local._schedulers.pop(str(tmp_path), None) @@ -135,9 +148,8 @@ def provider(): def test_scheduler_uses_shared_production_policy_for_poll_and_claim(monkeypatch): - from reflexio import durable_learning_policy - - monkeypatch.setattr(durable_learning_policy, "DISCOVERY_POLL_SECONDS", 2.0) + monkeypatch.setenv("REFLEXIO_DURABLE_LEARNING_POLL_SECONDS", "7.5") + monkeypatch.setenv("REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS", "45") monkeypatch.setattr( "reflexio.server.services.durable_learning.scheduler.worker_count", lambda: 1 ) @@ -150,5 +162,19 @@ def test_scheduler_uses_shared_production_policy_for_poll_and_claim(monkeypatch) "start_org", lambda org, lease: admitted.append((org, lease)) or True, ) - assert scheduler._run_once() == 2.0 - assert admitted == [("org", 300)] + assert scheduler._run_once() == 7.5 + assert admitted == [("org", 45)] + + +def test_invalid_policy_fails_before_constructing_a_worker(monkeypatch): + monkeypatch.setenv("REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS", "1.5") + monkeypatch.setattr( + "reflexio.server.services.durable_learning.scheduler.DurableLearningWorker", + lambda *_args, **_kwargs: pytest.fail( + "worker created before policy validation" + ), + ) + with pytest.raises(ValueError, match="LEASE_SECONDS"): + DurableLearningScheduler( + request_context_factory=_unused_context, org_ids_provider=lambda: [] + ) diff --git a/tests/server/test_env_utils.py b/tests/server/test_env_utils.py index d2fb95788..37d4b9ac5 100644 --- a/tests/server/test_env_utils.py +++ b/tests/server/test_env_utils.py @@ -2,6 +2,7 @@ import pytest +from reflexio.env_utils import env_truthy as shared_env_truthy from reflexio.server.env_utils import ( EnvBoolError, env_bool, @@ -13,6 +14,10 @@ ) +def test_legacy_parser_export_is_the_shared_parser() -> None: + assert env_truthy is shared_env_truthy + + def test_env_get_strips_values() -> None: assert env_get({"KEY": " value "}, "KEY") == "value" assert env_get({}, "KEY") == "" diff --git a/tests/test_durable_learning_policy.py b/tests/test_durable_learning_policy.py index c03527e13..4d929b36d 100644 --- a/tests/test_durable_learning_policy.py +++ b/tests/test_durable_learning_policy.py @@ -1,4 +1,4 @@ -"""Retiring discovery overrides cannot silently change a custom deployment.""" +"""Independent discovery and aggregation timing retain real operator control.""" import os import subprocess @@ -20,12 +20,32 @@ }, { "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS": " 2 ", - "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS": "300", + "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS": "300.0", }, ], ) -def test_retired_defaults_are_accepted_without_becoming_overrides(env): - validate_discovery_timing_env(env) +def test_discovery_defaults_and_blank_inputs_preserve_two_second_poll_and_lease(env): + from reflexio.durable_learning_policy import resolve_discovery_timing + + timing = resolve_discovery_timing(env) + assert (timing.poll_seconds, timing.lease_seconds) == (2.0, 300) + + +@pytest.mark.parametrize("poll,lease", [("0.05", "1"), ("10", "60"), ("7200", "900")]) +def test_custom_policies_are_independent_and_mapping_authoritative( + monkeypatch, poll, lease +): + from reflexio.durable_learning_policy import resolve_discovery_timing + + monkeypatch.setenv("REFLEXIO_DURABLE_LEARNING_POLL_SECONDS", "invalid-ambient") + env = { + "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS": poll, + "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS": lease, + } + timing = resolve_discovery_timing(env) + assert (timing.poll_seconds, timing.lease_seconds) == (float(poll), int(lease)) + assert resolve_discovery_timing({}).poll_seconds == 2.0 + assert env["REFLEXIO_DURABLE_LEARNING_POLL_SECONDS"] == poll @pytest.mark.parametrize( @@ -35,29 +55,72 @@ def test_retired_defaults_are_accepted_without_becoming_overrides(env): "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS", ], ) -@pytest.mark.parametrize("value", ["0", "-1", "0.05", "10", "nan", "inf", "invalid"]) -def test_custom_or_invalid_retired_timing_requires_migration(key, value): - with pytest.raises(ValueError, match=key): +@pytest.mark.parametrize( + "value", ["0", "-1", "nan", "inf", "private-invalid-input", "1e100"] +) +def test_invalid_discovery_timing_fails_with_value_free_errors(key, value): + with pytest.raises(ValueError, match=key) as error: validate_discovery_timing_env({key: value}) + assert "private-invalid-input" not in str(error.value) + + +@pytest.mark.parametrize("value", ["0.05", "1.5"]) +def test_claim_duration_requires_an_integral_number_of_seconds(value): + with pytest.raises(ValueError, match="LEASE_SECONDS"): + validate_discovery_timing_env( + {"REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS": value} + ) -def test_server_import_rejects_custom_timing_before_workers_start(tmp_path): +def test_discovery_wait_and_epoch_boundaries_are_enforced(): + from reflexio.durable_learning_policy import ( + MAX_DISCOVERY_LEASE_SECONDS, + resolve_discovery_timing, + ) + + timing = resolve_discovery_timing( + { + "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS": str(threading.TIMEOUT_MAX), + "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS": str(MAX_DISCOVERY_LEASE_SECONDS), + } + ) + assert timing.lease_seconds / 3 <= threading.TIMEOUT_MAX + with pytest.raises(ValueError, match="LEASE_SECONDS"): + resolve_discovery_timing( + { + "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS": str( + MAX_DISCOVERY_LEASE_SECONDS + 1 + ) + } + ) + with pytest.raises(ValueError, match="POLL_SECONDS"): + resolve_discovery_timing( + {"REFLEXIO_DURABLE_LEARNING_POLL_SECONDS": str(threading.TIMEOUT_MAX * 2)} + ) + + +@pytest.mark.parametrize( + "value,accepted", [("10", True), ("private-invalid-input", False)] +) +def test_server_import_validates_policy_before_workers_start(tmp_path, value, accepted): env = { **os.environ, - "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS": "10", + "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS": value, + "REFLEXIO_DURABLE_LEARNING_LEASE_SECONDS": "900", + "REFLEXIO_LOG_DIR": str(tmp_path), "LOCAL_STORAGE_PATH": str(tmp_path), } - result = subprocess.run( # noqa: S603 - fixed interpreter and literal import, no shell/input code. + result = subprocess.run( # noqa: S603 - fixed interpreter and literal import; no shell. [sys.executable, "-c", "import reflexio.server"], env=env, text=True, capture_output=True, timeout=30, ) - assert result.returncode != 0 - assert ( - "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS is retired; remove it" in result.stderr - ) + assert (result.returncode == 0) is accepted, result.stderr + if not accepted: + assert "REFLEXIO_DURABLE_LEARNING_POLL_SECONDS" in result.stderr + assert "private-invalid-input" not in result.stderr _AGGREGATION_TIMING_KEYS = ( diff --git a/tests/test_inference_launch.py b/tests/test_inference_launch.py new file mode 100644 index 000000000..204c7a322 --- /dev/null +++ b/tests/test_inference_launch.py @@ -0,0 +1,107 @@ +"""Endpoint ownership is shared by clients, CLI and container adapters.""" + +import subprocess +import sys +from pathlib import Path + +import pytest + +from reflexio.config_migration import ConfigurationMigrationError +from reflexio.inference_launch import ( + container_inference_launch, + inference_service_url, + ordinary_inference_launch, +) + + +def test_launcher_reuses_parser_without_importing_server_or_site_packages(): + package_root = Path(__file__).resolve().parents[1] + script = f""" +import sys +sys.path.insert(0, {str(package_root)!r}) +from reflexio.inference_launch import ordinary_inference_launch +from reflexio.env_utils import env_truthy +assert env_truthy(' ON ') +assert ordinary_inference_launch({{}}).mode == 'colocated' +assert not any(name == 'reflexio.server' or name.startswith('reflexio.server.') + for name in sys.modules) +""" + result = subprocess.run( # noqa: S603 — fixed interpreter and repository-owned probe + [sys.executable, "-I", "-S", "-c", script], + capture_output=True, + text=True, + timeout=15, + check=False, + ) + assert result.returncode == 0, result.stderr + + +@pytest.mark.parametrize( + "url", + [ + "https://remote.example/inference/", + "http://127.0.0.1:9999", + "http://localhost:9999", + ], +) +def test_explicit_endpoint_attaches_without_losing_its_address(url): + env = {"REFLEXIO_EMBEDDING_SERVICE_URL": url} + assert inference_service_url(env) == url.rstrip("/") + assert ordinary_inference_launch(env).mode == "attached" + assert ordinary_inference_launch(env).health_url is None + + +@pytest.mark.parametrize("legacy", ["true", "TrUe", "1", "yes", "on"]) +def test_disabling_without_an_endpoint_requires_migration(legacy): + with pytest.raises(ConfigurationMigrationError, match="explicitly"): + ordinary_inference_launch({"REFLEXIO_DISABLE_LOCAL_EMBEDDING_DAEMON": legacy}) + + +@pytest.mark.parametrize("legacy", ["", "false", "0", "no", "off"]) +def test_default_ownership_and_port_context(legacy): + env = {"REFLEXIO_DISABLE_LOCAL_EMBEDDING_DAEMON": legacy, "EMBEDDING_PORT": " "} + assert ordinary_inference_launch(env).health_url == "http://127.0.0.1:8072/health" + assert container_inference_launch(env).health_url == "http://127.0.0.1:8089/health" + env["EMBEDDING_PORT"] = "9999" + assert ordinary_inference_launch(env).health_url == "http://127.0.0.1:9999/health" + + +def test_custom_legacy_host_is_not_silently_dropped_or_printed(): + secret = "private-customer-host.example" + env = {"REFLEXIO_EMBEDDING_DAEMON_HOST": secret} + with pytest.raises(ConfigurationMigrationError) as exc: + inference_service_url(env) + assert secret not in str(exc.value) + assert "TIMEOUT_MS" in str(exc.value) + env.update( + REFLEXIO_EMBEDDING_SERVICE_URL="https://canonical.example", + REFLEXIO_DISABLE_LOCAL_EMBEDDING_DAEMON="true", + ) + assert inference_service_url(env) == "https://canonical.example" + assert ordinary_inference_launch(env).mode == "attached" + + +def test_query_child_remains_independent_of_ordinary_disable_switch(): + env = { + "REFLEXIO_API_QUERY_EMBEDDING_ENABLED": "true", + "REFLEXIO_QUERY_EMBEDDING_SERVICE_URL": "http://127.0.0.1:8093", + "REFLEXIO_DISABLE_LOCAL_EMBEDDING_DAEMON": "true", + } + plan = container_inference_launch(env) + assert plan.mode == "query" + assert plan.health_url == "http://127.0.0.1:8093/health/model" + env["REFLEXIO_QUERY_EMBEDDING_SERVICE_URL"] = "http://remote.example" + with pytest.raises(ConfigurationMigrationError, match="reserved loopback"): + container_inference_launch(env) + + +@pytest.mark.parametrize("url", ["http://127.0.0.1:8093", " http://127.0.0.1:8093/ "]) +def test_query_endpoint_normalization_matches_request_routing(url): + plan = container_inference_launch( + { + "REFLEXIO_API_QUERY_EMBEDDING_ENABLED": "true", + "REFLEXIO_QUERY_EMBEDDING_SERVICE_URL": url, + } + ) + assert plan.mode == "query" + assert plan.health_url == "http://127.0.0.1:8093/health/model"