diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml index 79cacb7cc1..19936e93fa 100644 --- a/.github/workflows/master.yml +++ b/.github/workflows/master.yml @@ -414,6 +414,9 @@ jobs: - name: Verify spawned worker lifecycle on Windows run: uv run --locked --no-sync pytest -q tests/builtin/runtime/test_artifact_processing.py + - name: Verify spawned worker logging on Windows + run: uv run --locked --no-sync pytest -q tests/e2e/test_worker_logging.py + - name: Verify bootstrap receipt persistence and failure recovery on Windows run: >- uv run --locked --no-sync pytest -q diff --git a/docs/en/development/observability.md b/docs/en/development/observability.md index 869c242fcb..e3ddf7e0f6 100644 --- a/docs/en/development/observability.md +++ b/docs/en/development/observability.md @@ -85,14 +85,11 @@ full database URLs, or sensitive filesystem paths. An allowed field still requir a redaction engine. Do not interpolate those values into the message or include them in a raw provider exception. Use `exc_info` only after reviewing the error's values and attached context for sensitive data. -The Server configures handlers and format. It injects the picklable `WorkerLogging` bootstrap into automatically -composed spawn workers through `worker_initializer`. Custom bindings own their launcher initialization. -Importing or running the Runtime as an SDK must not configure global logging. Keep Runtime observations independent -of Server settings, Prometheus, and OTel implementation types. - -Operational logs are best-effort. The Server handler isolates filter/formatter/write failures and emits one -content-free failure notification per handler. Do not use logs or sampled traces as a durable audit, billing, -or commit ledger. Request/trace/scope/source/worker identities do not belong in metric labels. +The Server configures handlers and format. It supplies `worker_initializer` for automatically composed spawn workers; file output prepares a picklable `WorkerLogging` bootstrap for each launch. Custom bindings own their launcher initialization. Importing or running the Runtime as an SDK must not configure global logging. Keep Runtime observations independent of Server settings, Prometheus, and OTel implementation types. + +Operational logs are best-effort. The Server handler isolates filter/formatter/write failures and emits one content-free failure notification per handler. File output uses `RotatingFileHandler` for rotation and retention, with `namer`/`rotator` and `gzip` for compression. The `worker_logging` context reuses the configured Server receiver or owns a worker-only sink for an embedded Server; it must outlive Runtime worker cleanup. Workers apply the field allowlist, identity binding, and formatting before sending. Only rendered UTF-8 output crosses the process boundary; the receiver must not rebind parent identity or serialize original extras. Before each spawn, the bootstrap's `prepare()` context creates a dedicated socketpair and yields the picklable child initializer. The launcher closes its sending end after spawn succeeds or cancellation cleanup completes; the receiver keeps reading until EOF. JSON-line envelopes preserve multiline console output. Malformed or incomplete records retire only their stream and are never echoed. Send waits are bounded. The receiver multiplexes streams into a bounded in-process queue feeding the native handlers. A full queue drops the record and reports a content-free failure. The Server stops Runtime and tracing before draining streams and calling the public `QueueListener.stop()` to finish queued output. The drain deadline bounds Worker-channel cleanup; file writes, compression, and close follow the native logging lifecycle without a separate deadline. Confirmed receiver discards increase the process-local drop counter independently of the once-only notification; clean EOF/reset does not. Logging starts no Manager or other helper process. + +Do not use logs or sampled traces as a durable audit, billing, or commit ledger. Request/trace/scope/source/worker identities do not belong in metric labels. ## Verify the behavior diff --git a/docs/en/docs/operate/configuration.md b/docs/en/docs/operate/configuration.md index 5aab440681..c1e8ffd13e 100644 --- a/docs/en/docs/operate/configuration.md +++ b/docs/en/docs/operate/configuration.md @@ -109,6 +109,11 @@ Server settings use the `POWERCONTEXT_SERVER_` prefix. | `POWERCONTEXT_SERVER_LOGGING_LEVEL` | `INFO` | Operational log level | | `POWERCONTEXT_SERVER_LOGGING_FORMAT` | `console` | `console` or structured `json` output | | `POWERCONTEXT_SERVER_LOGGING_ACCESS` | `true` | Log external HTTP and logical MCP request completion | +| `POWERCONTEXT_SERVER_LOGGING_FILE` | `false` | Write managed files instead of stdout; enabled by default for macOS/Windows personal services | +| `POWERCONTEXT_SERVER_LOGGING_DIRECTORY` | User data directory + `logs/` | Managed log directory | +| `POWERCONTEXT_SERVER_LOGGING_MAX_BYTES` | `20971520` | Positive size threshold for rotation | +| `POWERCONTEXT_SERVER_LOGGING_BACKUP_COUNT` | `10` | Positive maximum archive count per active file | +| `POWERCONTEXT_SERVER_LOGGING_COMPRESSION` | `gzip` | Archive compression: `gzip` or `none` | | `POWERCONTEXT_SERVER_METRICS_ENABLED` | `true` | Expose Prometheus metrics at `/metrics` | | `POWERCONTEXT_SERVER_METRICS_BACKGROUND_PORT` | `8001` | Background-role infrastructure listener port; uses `HTTP_HOST` and is disabled with metrics | | `POWERCONTEXT_SERVER_TRACING_ENABLED` | `false` | Enable span recording and OTLP export | diff --git a/docs/en/docs/operate/observability.md b/docs/en/docs/operate/observability.md index 0a17d0d7c1..b472af8e3a 100644 --- a/docs/en/docs/operate/observability.md +++ b/docs/en/docs/operate/observability.md @@ -15,6 +15,34 @@ The foreground Server writes operational logs to its process output. Configure and `POWERCONTEXT_SERVER_LOGGING_ACCESS` (default `true`). For a native personal service, run `powercontext service status` to find the journal selector or log paths. Docker users can inspect container logs. +### File output, rotation, and compression + +Enable file output with `POWERCONTEXT_SERVER_LOGGING_FILE=true`. The Server creates the directory and writes operational logs to files instead of stdout. Python's `RotatingFileHandler` defaults to rotation near 20 MiB and retains up to ten gzip archives per active file. Adjust the threshold, retention, and compression to fit the available storage and CPU budget. Restarting appends to the active file; no external rotation tool is required. + +| Environment variable | Default | Meaning | +| --- | --- | --- | +| `POWERCONTEXT_SERVER_LOGGING_FILE` | `false` | Write managed files instead of stdout | +| `POWERCONTEXT_SERVER_LOGGING_DIRECTORY` | User data directory + `logs/` | Directory for log files; respects `POWERCONTEXT_HOME` | +| `POWERCONTEXT_SERVER_LOGGING_MAX_BYTES` | `20971520` | Positive size threshold for rotation | +| `POWERCONTEXT_SERVER_LOGGING_BACKUP_COUNT` | `10` | Maximum archives per active file; must be at least one | +| `POWERCONTEXT_SERVER_LOGGING_COMPRESSION` | `gzip` | Archive compression: `gzip` or `none` | + +```bash +export POWERCONTEXT_SERVER_LOGGING_FILE=true +export POWERCONTEXT_SERVER_LOGGING_DIRECTORY=/var/lib/powercontext/logs +powercontext server run +``` + +Files are named by the existing Server process role: `all.log`, `api.log`, or `background.log`; explicit configuration without a role uses `server.log`. Workers share their owning Server's file, with execution identity recorded in each entry. Archives are named `all.log.1.gz`, `all.log.2.gz`, and so on, with `.1` being the newest archive. Without compression they use `.1`, `.2`, and so on. Changing compression does not remove archives from the previous policy. The active file remains uncompressed. The size threshold is approximate; one large record can exceed it. Retention applies per active file, not to unrelated files in the directory. `POWERCONTEXT_SERVER_LOGGING_FORMAT` selects readable text or JSON for either stdout or file output. + +Workers send already formatted, allowlisted records over dedicated socketpairs to their owning Server. Each channel is created before spawn, and only its sending end is passed to that Worker. The receiver does not listen on a network port or deserialize Python objects. JSON-line envelopes preserve multiline output. Worker identity and request/trace correlation are captured before sending. One receiver thread multiplexes the streams and one writer thread consumes a bounded in-process queue; logging starts no additional processes. Stream-only output keeps the standard stream handlers. + +Each file has an exclusive writer lock; a second Server with the same directory and role is rejected. Use separate directories for independent replicas. Stop workers before closing the logging owner. Shutdown drains Worker streams for at most one second, then waits for queued output through `QueueListener.stop()` before closing the handlers. File writes, compression, and close use native logging cleanup without a separate timeout. A stalled file sink can therefore delay Server process exit. Incomplete records are discarded without printing their contents. A failed Worker stream does not stop other streams. Send waits are bounded; a failed send retires that Worker's stream. The receiver handles up to 256 concurrent streams, and JSON-line envelopes are limited to 256 KiB after UTF-8 encoding. Oversized records, full queues, transport failures, and forced process exits can lose logs. Logs remain best-effort. + +macOS and Windows personal services enable managed file output by default. An explicit `POWERCONTEXT_SERVER_LOGGING_FILE=false` keeps stream output. Linux personal services keep journal output by default. Host-owned stdout/stderr files still capture startup diagnostics and direct stream writes; those files are outside the managed handler's rotation policy. `powercontext service status` reports the host's diagnostic directory or journal selector; a custom managed directory is selected through `POWERCONTEXT_SERVER_LOGGING_DIRECTORY`. + +An unavailable directory or an existing writer prevents file logging configuration. Once running, formatting, transport, queue, write, and compression failures use the same content-free `observability.log_failed` notification as stream failures. Completed gzip archives are published before the original file is removed. SDK Runtime use does not configure logging; embedding applications retain ownership. An embedded Server with file output can own a worker-only sink without reconfiguring the embedding application's loggers. + Use the response's `X-PowerContext-Request-ID` to correlate a failed request with diagnostics. The host guides also explain content-free recall and capture diagnostics; a missing context result can be valid rather than an outage. @@ -125,7 +153,7 @@ Runtime/Supervisor instance and reset when that owner is recreated; scrapes do n | `dependency_probe_duration_seconds` | `dependency`, `outcome`: actual probes, excluding cache hits | | `telemetry_collection_timestamp_seconds` | Last successful passive snapshot collection; `0` before collection | | `telemetry_collection_failures_total` | Failed passive collection attempts; the previous observation is retained | -| `telemetry_log_records_dropped_total` | Records that the process-owned log handler failed to deliver | +| `telemetry_log_records_dropped_total` | Observed records that the process-owned handler or Worker receiver failed to deliver; clean EOF/reset is not a loss, and unobserved losses are not counted | Detailed probe observations cover `database`, `inference.generation`, `inference.rerank`, `inference.decision`, and `inference.embedding`. The same `dependency_state` family also includes bounded aggregate checks for `runtime`, diff --git a/docs/en/rfcs/0046_observability_foundations.md b/docs/en/rfcs/0046_observability_foundations.md index 6d6354cb67..5ca15280d6 100644 --- a/docs/en/rfcs/0046_observability_foundations.md +++ b/docs/en/rfcs/0046_observability_foundations.md @@ -149,6 +149,8 @@ operation and its outcome. ## Logging boundary +> Implementation note: The Server supports optional file output with size-based rotation and gzip archives through standard-library logging. macOS/Windows personal services enable it by default. Shutdown bounds Worker-channel cleanup; native file I/O cleanup has no separate timeout. See [file logging](../docs/operate/observability.md#file-output-rotation-and-compression). + Server logging covers: - startup, readiness changes, and shutdown; diff --git a/docs/zh/development/observability.md b/docs/zh/development/observability.md index 8a6e401917..27c0d00142 100644 --- a/docs/zh/development/observability.md +++ b/docs/zh/development/observability.md @@ -86,13 +86,10 @@ Error classification 来自 `powercontext._logging.error_category`；`blocked` 这些值也不能拼入 message 或原样包含在 provider exception 中。 只有确认 error values 和关联上下文不含敏感数据后，才使用 `exc_info`。 -Server 配置 handlers 和 format，通过 `worker_initializer` 向自动构建的 spawn workers 注入可序列化的 -`WorkerLogging` bootstrap。Custom bindings 自己管理 launcher initialization。 -通过 SDK 导入或运行 Runtime 时，不得配置 global logging。 -Runtime observations 不依赖 Server settings、Prometheus 或 OTel implementation types。 +Server 配置 handlers 和 format，通过 `worker_initializer` 初始化自动构建的 spawn workers。文件输出为每次启动准备可序列化的 `WorkerLogging` bootstrap。Custom bindings 自己管理 launcher initialization。通过 SDK 导入或运行 Runtime 时，不得配置 global logging。Runtime observations 不依赖 Server settings、Prometheus 或 OTel implementation types。 + +Operational logs 保持 best-effort。Server handler 隔离 filter、formatter 和 write failures，每个 handler 最多输出一次无内容的 failure notification。文件输出由标准库 `RotatingFileHandler` 管理轮替和保留，通过 `namer`/`rotator` 与 `gzip` 压缩归档。`worker_logging` context 复用已配置的 Server receiver，或为嵌入式 Server 创建仅接收 Worker 日志的 sink；其退出必须晚于 Runtime 的 Worker 清理。Worker 在发送端完成字段过滤、identity 绑定和格式化，跨进程只发送格式化后的 UTF-8 内容；接收端不得重新绑定父进程 identity 或序列化原始 extras。每次 spawn 前，bootstrap 的 `prepare()` context 创建专用 socketpair，返回可序列化的子进程 initializer。launcher 在 spawn 完成或取消清理结束后关闭父进程持有的发送端，接收器持续读取到 EOF。日志通过 JSON 行封装，保留 console 输出中的换行。畸形或残缺记录只影响所属通道，不得回显原始内容。发送等待有上限。接收线程复用多个通道，通过有界的进程内队列交给原生 handler；队列满时丢弃该记录并报告无内容的失败通知。Server 先停止 Runtime 和 tracing，再排空日志通道，通过公开的 `QueueListener.stop()` 完成队列中的输出。排空期限约束 Worker 通道清理；文件写入、压缩和关闭使用原生 logging 生命周期，不另设超时。接收端确认丢弃的记录独立计入进程丢失计数，不受一次性通知限制；干净的 EOF/reset 不计数。日志处理不启动 Manager 或其他辅助进程。 -Operational logs 保持 best-effort。Server handler 隔离 filter、formatter 和 write failures， -每个 handler 最多输出一次无内容的 failure notification。 不要把 logs 或 sampled traces 当作 durable audit、billing 或 commit ledger。 Request、trace、scope、source 和 worker identities 不作为 metric labels。 diff --git a/docs/zh/docs/operate/configuration.md b/docs/zh/docs/operate/configuration.md index dd2f1177b4..62c9d7334d 100644 --- a/docs/zh/docs/operate/configuration.md +++ b/docs/zh/docs/operate/configuration.md @@ -100,6 +100,11 @@ Server 配置使用 `POWERCONTEXT_SERVER_` 前缀。 | `POWERCONTEXT_SERVER_LOGGING_LEVEL` | `INFO` | operational log 级别 | | `POWERCONTEXT_SERVER_LOGGING_FORMAT` | `console` | `console` 或结构化 `json` 输出 | | `POWERCONTEXT_SERVER_LOGGING_ACCESS` | `true` | 记录外部 HTTP 和逻辑 MCP request completion | +| `POWERCONTEXT_SERVER_LOGGING_FILE` | `false` | 写入托管文件以替代 stdout；macOS/Windows 个人服务默认开启 | +| `POWERCONTEXT_SERVER_LOGGING_DIRECTORY` | 用户数据目录下的 `logs/` | 托管日志目录 | +| `POWERCONTEXT_SERVER_LOGGING_MAX_BYTES` | `20971520` | 按大小轮替的阈值，必须为正数 | +| `POWERCONTEXT_SERVER_LOGGING_BACKUP_COUNT` | `10` | 每个活动文件最多保留的归档数，至少为一 | +| `POWERCONTEXT_SERVER_LOGGING_COMPRESSION` | `gzip` | 归档压缩方式：`gzip` 或 `none` | | `POWERCONTEXT_SERVER_METRICS_ENABLED` | `true` | 在 `/metrics` 暴露 Prometheus metrics | | `POWERCONTEXT_SERVER_METRICS_BACKGROUND_PORT` | `8001` | Background role 的 infrastructure listener 端口；使用 `HTTP_HOST`，随 metrics 一起关闭 | | `POWERCONTEXT_SERVER_TRACING_ENABLED` | `false` | 启用 span recording 和 OTLP export | diff --git a/docs/zh/docs/operate/observability.md b/docs/zh/docs/operate/observability.md index ba47d97734..02933bf89b 100644 --- a/docs/zh/docs/operate/observability.md +++ b/docs/zh/docs/operate/observability.md @@ -14,6 +14,34 @@ description: 通过日志定位请求，并查看 Server 指标与 Trace。 `POWERCONTEXT_SERVER_LOGGING_FORMAT`（`console` 或 `json`）和 `POWERCONTEXT_SERVER_LOGGING_ACCESS`（默认 `true`）。 原生个人服务通过 `powercontext service status` 查看 journal 查询条件或日志路径；Docker 部署查看容器日志。 +### 文件输出、轮替与压缩 + +设置 `POWERCONTEXT_SERVER_LOGGING_FILE=true` 启用文件输出，Server 自动创建目录，将运行日志写入文件以替代 stdout。标准库 `RotatingFileHandler` 默认在接近 20 MiB 时轮替，每个活动文件最多保留十个 gzip 归档。轮替大小、保留数量和压缩方式可按存储及 CPU 预算调整；重启后继续追加活动文件，无需外部轮替工具。 + +| 环境变量 | 默认值 | 含义 | +| --- | --- | --- | +| `POWERCONTEXT_SERVER_LOGGING_FILE` | `false` | 写入托管文件以替代 stdout | +| `POWERCONTEXT_SERVER_LOGGING_DIRECTORY` | 用户数据目录下的 `logs/` | 日志目录，遵循 `POWERCONTEXT_HOME` | +| `POWERCONTEXT_SERVER_LOGGING_MAX_BYTES` | `20971520` | 按大小轮替的阈值，必须为正数 | +| `POWERCONTEXT_SERVER_LOGGING_BACKUP_COUNT` | `10` | 每个活动文件最多保留的归档数，至少为一 | +| `POWERCONTEXT_SERVER_LOGGING_COMPRESSION` | `gzip` | 归档压缩方式：`gzip` 或 `none` | + +```bash +export POWERCONTEXT_SERVER_LOGGING_FILE=true +export POWERCONTEXT_SERVER_LOGGING_DIRECTORY=/var/lib/powercontext/logs +powercontext server run +``` + +文件按已有 Server 进程角色命名为 `all.log`、`api.log` 或 `background.log`；显式配置但未指定角色时使用 `server.log`。Worker 共用所属 Server 的文件，每条日志保留执行身份。归档依次为 `all.log.1.gz`、`all.log.2.gz`，`.1` 为最新归档；关闭压缩时归档后缀为 `.1`、`.2` 等，切换压缩策略不会删除旧策略的归档。活动文件保持未压缩。大小阈值是近似值，单条大日志可能超过阈值。保留数量仅约束对应活动文件的归档，不清理目录内的其他文件。`POWERCONTEXT_SERVER_LOGGING_FORMAT` 选择可读文本或 JSON 格式，对 stdout 和文件输出均有效。 + +Worker 在发送端完成格式化和字段白名单过滤，通过各自专用的 socketpair 将日志发送给所属 Server。每条通道在 spawn 前创建，仅发送端传递给对应 Worker。接收器不监听网络端口，不反序列化 Python 对象。JSON 行封装保留输出中的换行，Worker 身份及 request/trace 关联在发送前确定。一个接收线程复用多个通道，一个写入线程消费有界的进程内队列；日志处理不额外创建进程。仅输出标准流时继续使用标准 stream handler。 + +每个文件使用独占写入锁，相同目录和角色的第二个 Server 会被拒绝；独立副本应使用不同目录。关闭日志前先停止 Worker。关闭时最多等待一秒排空 Worker 日志通道，再通过 `QueueListener.stop()` 完成队列中的输出，最后关闭 handler；文件写入、压缩和关闭使用原生 logging 清理，不另设超时；文件 sink 阻塞时可能延迟 Server 进程退出。残缺记录直接丢弃，不打印其内容；单个 Worker 通道故障不影响其他通道。发送等待有上限，发送失败后关闭该 Worker 的日志通道。接收器最多处理 256 条并发通道，单条 JSON 行封装在 UTF-8 编码后最多 256 KiB。超大记录、队列满、传输故障或进程被强制退出时可能丢失日志，日志仍为 best-effort。 + +macOS 和 Windows 个人服务默认启用托管文件输出，显式设置 `POWERCONTEXT_SERVER_LOGGING_FILE=false` 可保留标准流输出。Linux 个人服务默认继续使用 journal。宿主管理的 stdout/stderr 文件仍接收启动诊断和直接的标准流写入，不受托管 handler 的轮替策略管理。`powercontext service status` 显示宿主诊断日志目录或 journal 查询条件；自定义托管目录由 `POWERCONTEXT_SERVER_LOGGING_DIRECTORY` 指定。 + +目录不可用或已有写入者时，文件日志配置失败。运行期间的格式化、传输、入队、写入及压缩失败均使用不含内容的 `observability.log_failed` 通知。gzip 归档完整发布后才删除原始文件。通过 SDK 使用 Runtime 不配置日志，仍由嵌入应用管理。嵌入式 Server 启用文件输出时可以持有仅用于 Worker 的日志接收器，不重新配置嵌入应用的 logger。 + 用响应中的 `X-PowerContext-Request-ID` 关联失败请求与诊断记录。宿主指南还说明不含内容的召回与采集诊断； 没有上下文结果可能是正常情况，不一定代表服务故障。 @@ -118,7 +146,7 @@ API 与后台进程同机部署时应使用不同端口。设置 `POWERCONTEXT_S | `dependency_probe_duration_seconds` | `dependency`、`outcome`：实际探测耗时，不包含缓存命中 | | `telemetry_collection_timestamp_seconds` | 最近成功复制被动快照的时间；首次采集前为 `0` | | `telemetry_collection_failures_total` | 被动采集失败次数；失败保留先前观察 | -| `telemetry_log_records_dropped_total` | 进程自有日志 handler 未能交付的记录数 | +| `telemetry_log_records_dropped_total` | 进程自有 handler 或 Worker 接收端确认未能交付的记录数；干净的 EOF/reset 不计入丢失，未观测到的丢失不计数 | 详细探测观察覆盖 `database`、`inference.generation`、`inference.rerank`、`inference.decision` 和 `inference.embedding`。同一 `dependency_state` 指标还包含有界总体检查：`runtime`、 diff --git a/docs/zh/rfcs/0046_observability_foundations.md b/docs/zh/rfcs/0046_observability_foundations.md index 5f1f157125..66464c26e6 100644 --- a/docs/zh/rfcs/0046_observability_foundations.md +++ b/docs/zh/rfcs/0046_observability_foundations.md @@ -144,6 +144,8 @@ outcome 的含义保持一致。 ## Logging boundary +> 实现补充：Server 支持使用标准库 logging 提供可选文件输出、按大小轮替和 gzip 归档，macOS/Windows 个人服务默认启用。停机时限约束 Worker 通道清理，原生文件 I/O 清理不另设超时。配置见[文件日志](../docs/operate/observability.md#文件输出轮替与压缩)。 + Server logging 覆盖： - startup、readiness change 和 shutdown； diff --git a/src/powercontext/builtin/runtime/artifact_processing.py b/src/powercontext/builtin/runtime/artifact_processing.py index a6c42932c8..b1fdaffd36 100644 --- a/src/powercontext/builtin/runtime/artifact_processing.py +++ b/src/powercontext/builtin/runtime/artifact_processing.py @@ -25,7 +25,7 @@ import traceback as traceback_module from collections import deque from collections.abc import Awaitable, Callable, Iterator, Sequence -from contextlib import contextmanager, nullcontext, suppress +from contextlib import ExitStack, contextmanager, nullcontext, suppress from dataclasses import dataclass, field, replace from datetime import UTC, datetime, timedelta from enum import StrEnum @@ -70,6 +70,7 @@ ArtifactProcessingWorkerLauncher, ArtifactProcessingWorkerOutcome, WorkerEntrypoint, + WorkerInitializerPreparation, processing_error_code, ) from powercontext.builtin.runtime.protocols import RuntimeSpan, RuntimeTracing @@ -218,7 +219,7 @@ def __init__( self, entrypoint: WorkerEntrypoint, *, - initializer: Callable[[], None] | None = None, + initializer: Callable[[], None] | WorkerInitializerPreparation | None = None, entrypoint_wrapper: Callable[[WorkerEntrypoint], WorkerEntrypoint] | None = None, collect_model_usage: bool = False, ) -> None: @@ -228,41 +229,45 @@ def __init__( self._collect_model_usage = collect_model_usage async def start(self, assignment: ArtifactProcessingWorkAssignment) -> ArtifactProcessingWorkerHandle: - context = multiprocessing.get_context("spawn") - receiver, sender = context.Pipe(duplex=False) - entrypoint = self._entrypoint - if self._entrypoint_wrapper is not None: - with suppress(Exception): - entrypoint = self._entrypoint_wrapper(entrypoint) - process = _OwnedSpawnProcess( - target=_run_spawned_worker, - args=(sender, entrypoint, assignment, self._initializer, self._collect_model_usage), - name=f"powercontext-artifact-worker-{assignment.worker_id}", - daemon=False, - ) - owner = _SpawnStartOwner() - start_task = asyncio.create_task( - asyncio.to_thread(_start_owned_spawn_process, process, owner), - name=f"powercontext-artifact-worker-start-{assignment.worker_id}", - ) - try: - await asyncio.shield(start_task) - except asyncio.CancelledError: - cleanup_task = asyncio.create_task( - _cleanup_cancelled_spawn_start(start_task, process, owner, receiver, sender), - name=f"powercontext-artifact-worker-cleanup-{assignment.worker_id}", + with ExitStack() as launch_resources: + initializer = self._initializer + if isinstance(initializer, WorkerInitializerPreparation): + initializer = launch_resources.enter_context(initializer.prepare()) + context = multiprocessing.get_context("spawn") + receiver, sender = context.Pipe(duplex=False) + entrypoint = self._entrypoint + if self._entrypoint_wrapper is not None: + with suppress(Exception): + entrypoint = self._entrypoint_wrapper(entrypoint) + process = _OwnedSpawnProcess( + target=_run_spawned_worker, + args=(sender, entrypoint, assignment, initializer, self._collect_model_usage), + name=f"powercontext-artifact-worker-{assignment.worker_id}", + daemon=False, ) - await _complete_cleanup(cleanup_task) - raise - except BaseException: - cleanup_task = asyncio.create_task( - _cleanup_failed_spawn_start(owner, receiver, sender), - name=f"powercontext-artifact-worker-cleanup-{assignment.worker_id}", + owner = _SpawnStartOwner() + start_task = asyncio.create_task( + asyncio.to_thread(_start_owned_spawn_process, process, owner), + name=f"powercontext-artifact-worker-start-{assignment.worker_id}", ) - await _complete_cleanup(cleanup_task) - raise - sender.close() - return _SpawnedWorkerHandle(process, receiver) + try: + await asyncio.shield(start_task) + except asyncio.CancelledError: + cleanup_task = asyncio.create_task( + _cleanup_cancelled_spawn_start(start_task, process, owner, receiver, sender), + name=f"powercontext-artifact-worker-cleanup-{assignment.worker_id}", + ) + await _complete_cleanup(cleanup_task) + raise + except BaseException: + cleanup_task = asyncio.create_task( + _cleanup_failed_spawn_start(owner, receiver, sender), + name=f"powercontext-artifact-worker-cleanup-{assignment.worker_id}", + ) + await _complete_cleanup(cleanup_task) + raise + sender.close() + return _SpawnedWorkerHandle(process, receiver) async def _cleanup_cancelled_spawn_start( diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index f02631453f..731021f170 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -178,7 +178,11 @@ from powercontext.builtin.runtime.family_processing import FAMILY_BINDINGS, FamilyWorkerSpec, run_family_worker from powercontext.builtin.runtime.memory_write_gate import build_memory_write_gate from powercontext.builtin.runtime.models import DreamOperationCapability, MemorySearchMode, RuntimeCapabilities -from powercontext.builtin.runtime.processing_contracts import ArtifactProcessingModelUsageDelta, WorkerEntrypoint +from powercontext.builtin.runtime.processing_contracts import ( + ArtifactProcessingModelUsageDelta, + WorkerEntrypoint, + WorkerInitializerPreparation, +) from powercontext.builtin.runtime.processing_discovery import SourceProcessingPendingProvider, enabled_profile_scopes from powercontext.builtin.runtime.processing_registry import ( canonical_processing_manifest, @@ -514,7 +518,7 @@ async def open_builtin_runtime( artifact_processing_bindings: Sequence[ArtifactProcessingBinding] = (), worker_security: dict[str, Any] | None = None, processing_observer: Callable[[str, str, float], None] | None = None, - worker_initializer: Callable[[], None] | None = None, + worker_initializer: Callable[[], None] | WorkerInitializerPreparation | None = None, worker_entrypoint_wrapper: Callable[[WorkerEntrypoint], WorkerEntrypoint] | None = None, scheduled_source_runner: ScheduledSourceRunner | None = None, scheduled_experience_runner: ScheduledExperienceRunner | None = None, @@ -530,9 +534,11 @@ async def open_builtin_runtime( An explicit ``recall_effort_sink`` replaces that recorder rather than adding a second observation. A disabled gate invokes neither recorder nor sink. - ``worker_initializer`` is an optional picklable bootstrap for automatically - composed spawn workers. Custom bindings own their launchers and initialization. - The default does not configure process logging. + ``worker_initializer`` is an optional bootstrap for automatically composed spawn + workers. A ``WorkerInitializerPreparation`` can own per-launch resources through + a context manager yielding the picklable child bootstrap. Its resources remain + open until spawn or cancellation cleanup finishes. Custom bindings own their + launchers and initialization. The default does not configure process logging. """ async with AsyncExitStack() as resources: @@ -920,7 +926,7 @@ def _artifact_processing_bindings( # noqa: C901 - validate and assemble one reg injected_pipelines: Mapping[str, object | None] | None = None, injected_memory_write_gate: MemoryWriteGate | None = None, worker_security: dict[str, Any] | None = None, - worker_initializer: Callable[[], None] | None = None, + worker_initializer: Callable[[], None] | WorkerInitializerPreparation | None = None, worker_entrypoint_wrapper: Callable[[WorkerEntrypoint], WorkerEntrypoint] | None = None, worker_model_usage_observer: Callable[[ArtifactProcessingModelUsageDelta | None], None] | None = None, source_registry: SourceDefinitionRegistry = BUILTIN_SOURCE_REGISTRY, diff --git a/src/powercontext/builtin/runtime/processing_contracts.py b/src/powercontext/builtin/runtime/processing_contracts.py index cd55503845..11dbb713a1 100644 --- a/src/powercontext/builtin/runtime/processing_contracts.py +++ b/src/powercontext/builtin/runtime/processing_contracts.py @@ -17,9 +17,10 @@ from __future__ import annotations from collections.abc import Callable +from contextlib import AbstractContextManager from dataclasses import dataclass from enum import StrEnum -from typing import Protocol +from typing import Protocol, runtime_checkable from powercontext.builtin.persistence.supervision import ArtifactProcessingFence @@ -130,6 +131,16 @@ async def start(self, assignment: ArtifactProcessingWorkAssignment) -> ArtifactP WorkerEntrypoint = Callable[[ArtifactProcessingWorkAssignment], ArtifactProcessingWorkerCompletion | None] +@runtime_checkable +class WorkerInitializerPreparation(Protocol): + """Own per-launch resources through spawn or cancellation cleanup. + + The context yields a picklable callable that runs inside the new child. + """ + + def prepare(self) -> AbstractContextManager[Callable[[], None]]: ... + + def processing_error_code(exception_type: type[BaseException] | str, error_code: str = "") -> str: """Expose only known categories, never exception messages or custom codes.""" @@ -165,5 +176,6 @@ def processing_error_code(exception_type: type[BaseException] | str, error_code: "ArtifactProcessingWorkerLauncher", "ArtifactProcessingWorkerOutcome", "WorkerEntrypoint", + "WorkerInitializerPreparation", "processing_error_code", ] diff --git a/src/powercontext/server/_log_transport.py b/src/powercontext/server/_log_transport.py new file mode 100644 index 0000000000..9aef94dc48 --- /dev/null +++ b/src/powercontext/server/_log_transport.py @@ -0,0 +1,203 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Dedicated Worker log sockets multiplexed by their owning Server.""" + +from __future__ import annotations + +import json +import selectors +import socket +import time +from collections.abc import Callable, Iterator +from contextlib import ExitStack, contextmanager, suppress +from queue import Empty, Queue +from threading import Event, Lock, Thread + +_MAX_RECORD_BYTES = 256 * 1024 +# Stay below Windows select()'s socket limit, including the wakeup socket. +_MAX_CHANNELS = 256 +_DRAIN_TIMEOUT = 1.0 + + +class LogSender: + """Send JSON lines; a partial send retires only this Worker's socket.""" + + def __init__(self, channel: socket.socket) -> None: + self._channel: socket.socket | None = channel + channel.settimeout(0.05) + + def send(self, message: str) -> None: + # JSON quoting preserves multiline console messages as a single transport record. + payload = json.dumps(message, ensure_ascii=False).encode("utf-8") + b"\n" + if len(payload) > _MAX_RECORD_BYTES: + raise ValueError("Worker log record exceeds the transport limit") # noqa: TRY003 + if self._channel is None: + raise OSError("Worker log stream is unavailable") # noqa: TRY003 + try: + self._channel.sendall(payload) + except OSError: + self.close() + raise + + def close(self) -> None: + if self._channel is not None: + self._channel.close() + self._channel = None + + +class LogReceiver: + """Multiplex dedicated Worker sockets without owning task execution.""" + + def __init__( + self, deliver: Callable[[str], None], on_failure: Callable[[], None], on_drop: Callable[[int], None] + ) -> None: + self._deliver = deliver + self._on_failure = on_failure + self._on_drop = on_drop + self._stopped = Event() + self._lock = Lock() + self._pending: Queue[socket.socket] = Queue(maxsize=_MAX_CHANNELS) + self._buffers: dict[socket.socket, bytearray] = {} + self._resources = ExitStack() + try: + self._selector = self._resources.enter_context(selectors.DefaultSelector()) + self._wake_read, self._wake_write = socket.socketpair() + self._resources.callback(self._wake_read.close) + self._resources.callback(self._wake_write.close) + self._wake_read.setblocking(False) + self._wake_write.setblocking(False) + self._selector.register(self._wake_read, selectors.EVENT_READ) + self._thread = Thread(target=self._run, name="powercontext-worker-logs", daemon=True) + self._thread.start() + except BaseException: + self._resources.close() + raise + + @contextmanager + def channel(self) -> Iterator[socket.socket]: + """Create before spawn and close the parent's sending end after spawn settles.""" + + receiver, sender = socket.socketpair() + try: + receiver.setblocking(False) + with self._lock: + if self._stopped.is_set(): + raise RuntimeError("Server log receiver is closed") # noqa: TRY003, TRY301 + self._pending.put_nowait(receiver) + except BaseException: + receiver.close() + sender.close() + raise + self._wake() + try: + yield sender + finally: + sender.close() + + def _wake(self) -> None: + with suppress(OSError): + self._wake_write.send(b"\0") + + def close(self) -> bool: + with self._lock: + self._stopped.set() + self._wake() + self._thread.join(timeout=_DRAIN_TIMEOUT + 0.5) + if self._thread.is_alive(): + self._on_failure() + return False + return True + + def _register_pending(self) -> None: + while True: + try: + connection = self._pending.get_nowait() + except Empty: + return + if len(self._buffers) >= _MAX_CHANNELS: + connection.close() + self._on_failure() + continue + try: + self._selector.register(connection, selectors.EVENT_READ) + self._buffers[connection] = bytearray() + except Exception: + connection.close() + self._on_failure() + + def _close_channel(self, connection: socket.socket) -> None: + buffer = self._buffers.pop(connection) + if buffer: + self._on_drop(buffer.count(b"\n") + int(not buffer.endswith(b"\n"))) + self._selector.unregister(connection) + connection.close() + + def _read(self, connection: socket.socket) -> None: + buffer = self._buffers[connection] + try: + data = connection.recv(64 * 1024) + if not data: + self._close_channel(connection) + return + buffer.extend(data) + self._decode(buffer) + except BlockingIOError: + return + except (ConnectionResetError, ConnectionAbortedError): + # Windows reports terminated peers as resets rather than EOF. + self._close_channel(connection) + except Exception: + self._on_failure() + self._close_channel(connection) + + def _decode(self, buffer: bytearray) -> None: + while (newline := buffer.find(b"\n")) != -1: + if newline + 1 > _MAX_RECORD_BYTES: + raise ValueError("Worker log record exceeds the transport limit") # noqa: TRY003 + message = json.loads(buffer[:newline]) + if not isinstance(message, str): + raise TypeError("Worker log record must contain formatted text") # noqa: TRY003 + del buffer[: newline + 1] + self._deliver(message) + if len(buffer) >= _MAX_RECORD_BYTES: + raise ValueError("Worker log record exceeds the transport limit") # noqa: TRY003 + + def _run(self) -> None: + deadline: float | None = None + try: + while True: + self._register_pending() + if self._stopped.is_set() and deadline is None: + deadline = time.monotonic() + _DRAIN_TIMEOUT + self._register_pending() + if deadline is not None and (not self._buffers or time.monotonic() >= deadline): + break + for key, _ in self._selector.select(timeout=0.05): + connection = key.fileobj + if connection is self._wake_read: + with suppress(BlockingIOError): + self._wake_read.recv(1024) + elif isinstance(connection, socket.socket): + self._read(connection) + except Exception: + self._on_failure() + finally: + # Closing admission before cleanup also covers an unexpected receiver failure. + with self._lock: + self._stopped.set() + self._register_pending() + for connection in list(self._buffers): + self._close_channel(connection) + self._resources.close() diff --git a/src/powercontext/server/cli.py b/src/powercontext/server/cli.py index 71e54787b8..de2dfbfcd1 100644 --- a/src/powercontext/server/cli.py +++ b/src/powercontext/server/cli.py @@ -56,7 +56,7 @@ from powercontext.server.database_migration import app as database_migration_app from powercontext.server.factory import _ServerReadinessProbe, create_background_metrics_app, create_server_app from powercontext.server.identity import open_server_identity_repository -from powercontext.server.logging import WorkerLogging, configure_server_logging +from powercontext.server.logging import configure_server_logging, shutdown_server_logging, worker_logging from powercontext.server.metrics import ServerMetrics from powercontext.server.processing_security import build_worker_security from powercontext.server.settings import ( @@ -411,23 +411,22 @@ def _run_configured_server(settings: ServerSettings) -> None: """Run one already-validated configuration in the current process.""" configure_server_logging(settings.logging, role=settings.runtime.artifact_processing_role) - tracing = configure_server_tracing(settings.tracing, role=settings.runtime.artifact_processing_role) try: - write_inference_capability_notice( - generation_model=settings.inference.generation_model, - embedding_model=settings.inference.embedding_model, - ) - if settings.runtime.artifact_processing_role == "background": - _run_background(settings, tracing) - return - application = create_server_app(settings=settings, tracing=tracing) - _run_server( - application, - host=settings.http.host, - port=settings.http.port, - ) + tracing = configure_server_tracing(settings.tracing, role=settings.runtime.artifact_processing_role) + try: + write_inference_capability_notice( + generation_model=settings.inference.generation_model, + embedding_model=settings.inference.embedding_model, + ) + if settings.runtime.artifact_processing_role == "background": + _run_background(settings, tracing) + return + application = create_server_app(settings=settings, tracing=tracing) + _run_server(application, host=settings.http.host, port=settings.http.port) + finally: + tracing.shutdown() finally: - tracing.shutdown() + shutdown_server_logging() def _friendly_bad_parameter(error: ValidationError) -> typer.BadParameter: @@ -482,6 +481,7 @@ async def _run_background_async(settings: ServerSettings, tracing: Any) -> None: installed.append(signum) try: async with AsyncExitStack() as resources: + logging_bootstrap = resources.enter_context(worker_logging(settings.logging, role="background")) principal = ( PrincipalRef(type="service", id="server-token", description="PowerContext static bearer") if settings.auth.token is not None @@ -506,7 +506,7 @@ async def _run_background_async(settings: ServerSettings, tracing: Any) -> None: model_usage_observer=None if metrics is None else metrics.observe_model_usage, worker_model_usage_observer=None if metrics is None else metrics.observe_worker_model_usage, processing_observer=None if metrics is None else metrics.observe_processing, - worker_initializer=WorkerLogging(settings.logging), + worker_initializer=logging_bootstrap, worker_entrypoint_wrapper=WorkerTracing(settings.tracing).wrap, ) ) diff --git a/src/powercontext/server/factory.py b/src/powercontext/server/factory.py index af56ad2ee7..b8577b287a 100644 --- a/src/powercontext/server/factory.py +++ b/src/powercontext/server/factory.py @@ -81,7 +81,7 @@ from powercontext.server.dream_access import DreamAccess from powercontext.server.identity import ServerIdentityRepository from powercontext.server.info import server_info -from powercontext.server.logging import WorkerLogging +from powercontext.server.logging import worker_logging from powercontext.server.mcp import mount_mcp from powercontext.server.metrics import CONTENT_TYPE_LATEST, HttpMetricsMiddleware, ServerMetrics from powercontext.server.middleware import AuthenticationMiddleware @@ -185,6 +185,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: ) cursor_secret = resolve_cursor_secret(config.database, configured_cursor_secret) async with AsyncExitStack() as resources: + logging_bootstrap = resources.enter_context( + worker_logging(resolved.logging, role=resolved.runtime.artifact_processing_role) + ) active_access_control = configured_access_control if active_access_control is None and resolved.access.mode == "enforced": active_access_control = await resources.enter_async_context( @@ -240,7 +243,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: scheduled_source_runner=scheduled_source_runner, scheduled_experience_runner=scheduled_experience_runner, worker_security=worker_security, - worker_initializer=WorkerLogging(resolved.logging), + worker_initializer=logging_bootstrap, worker_entrypoint_wrapper=WorkerTracing(resolved.tracing).wrap, scheduled_profile_runner=_scheduled_profile_runner( resolved, diff --git a/src/powercontext/server/logging.py b/src/powercontext/server/logging.py index 08527a1d3a..57aabaa25c 100644 --- a/src/powercontext/server/logging.py +++ b/src/powercontext/server/logging.py @@ -16,20 +16,31 @@ from __future__ import annotations +import atexit +import gzip import json import logging import logging.config +import shutil +import socket import sys -from contextlib import suppress +import tempfile +from collections.abc import Iterator +from contextlib import ExitStack, contextmanager, suppress from dataclasses import dataclass from datetime import UTC, datetime +from logging.handlers import QueueHandler, QueueListener, RotatingFileHandler +from pathlib import Path +from queue import Full, Queue from threading import Lock from typing import TYPE_CHECKING, Any, TextIO +from filelock import FileLock from opentelemetry import trace from typing_extensions import override from powercontext._logging import current_log_context, error_category +from powercontext.server._log_transport import LogReceiver, LogSender from powercontext.server.context import current_request_id from powercontext.server.settings import ServerLoggingConfig from powercontext.server.telemetry import service_resource @@ -203,18 +214,16 @@ def filter(self, record: logging.LogRecord) -> bool: def log_record_failures() -> int: - """Return process-local records the configured handler could not deliver.""" + """Return observed process-local log records that could not be delivered.""" with _LOG_FAILURE_LOCK: return _LOG_FAILURES -class _OperationalStreamHandler(_StreamHandler): +class _FailureIsolatedHandler(logging.Handler): """Report one bounded failure without Python logging's raw-record stderr dump.""" - def __init__(self, stream: TextIO | None = None) -> None: - super().__init__(stream) - self._failure_reported = False + _failure_reported = False @override def handle(self, record: logging.LogRecord) -> bool: @@ -226,13 +235,147 @@ def handle(self, record: logging.LogRecord) -> bool: @override def handleError(self, record: logging.LogRecord) -> None: + self._record_failures() + + def _record_failures(self, count: int = 1) -> None: global _LOG_FAILURES with _LOG_FAILURE_LOCK: - _LOG_FAILURES += 1 - if not self._failure_reported: + _LOG_FAILURES += count + self._notify_failure() + + def _notify_failure(self) -> None: + with _LOG_FAILURE_LOCK: + if self._failure_reported: + return self._failure_reported = True - with suppress(Exception): - sys.stderr.write('{"event":"observability.log_failed","outcome":"failure"}\n') + with suppress(Exception): + sys.stderr.write('{"event":"observability.log_failed","outcome":"failure"}\n') + + +class _OperationalStreamHandler(_FailureIsolatedHandler, _StreamHandler): + pass + + +class _OperationalRotatingFileHandler(_FailureIsolatedHandler, RotatingFileHandler): + pass + + +def _gzip_archive(source: str, destination: str) -> None: + """Publish complete archives before removing their uncompressed source.""" + + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile(dir=Path(destination).parent, delete=False) as output: + temporary = Path(output.name) + with Path(source).open("rb") as input_stream, gzip.GzipFile(fileobj=output, mode="wb") as compressed: + shutil.copyfileobj(input_stream, compressed) + temporary.replace(destination) + Path(source).unlink() + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +class _OperationalSocketHandler(_FailureIsolatedHandler): + def __init__(self, channel: socket.socket) -> None: + super().__init__() + self._sender = LogSender(channel) + + @override + def emit(self, record: logging.LogRecord) -> None: + self._sender.send(self.format(record)) + + @override + def close(self) -> None: + self.acquire() + try: + self._sender.close() + finally: + self.release() + super().close() + + +class _DrainingQueueListener(QueueListener): + queue: Queue[logging.LogRecord | None] + _sentinel = None + + @override + def enqueue_sentinel(self) -> None: + # The default put_nowait can fail when a bounded queue is full at shutdown. + self.queue.put(self._sentinel) + + +class _OperationalQueueHandler(_FailureIsolatedHandler, QueueHandler): + """Own native file handlers and receive independent Worker log streams.""" + + def __init__(self, config: ServerLoggingConfig, *, role: str) -> None: + self.config = config + self._resources = ExitStack() + self._listener: QueueListener | None = None + self._receiver: LogReceiver | None = None + self._output_queue: Queue[logging.LogRecord | None] = Queue(maxsize=1024) + try: + config.directory.mkdir(mode=0o700, parents=True, exist_ok=True) + filename = config.directory / f"{role}.log" + lock = FileLock(str(filename) + ".lock", thread_local=False) + lock.acquire(timeout=0) + self._resources.callback(lock.release) + file_handler = _OperationalRotatingFileHandler( + filename, maxBytes=config.max_bytes, backupCount=config.backup_count, encoding="utf-8" + ) + self._resources.callback(file_handler.close) + if config.compression == "gzip": + file_handler.namer = lambda name: name + ".gz" + file_handler.rotator = _gzip_archive + # logging.shutdown closes handlers in reverse registration order. + super().__init__(self._output_queue) + listener = _DrainingQueueListener(self._output_queue, file_handler) + listener.start() + self._listener = listener + self._receiver = LogReceiver(self._receive, self._notify_failure, self._record_failures) + atexit.register(self.close) + except BaseException: + if self._listener is None: + self._resources.close() + else: + self.close() + raise + + @contextmanager + def worker_channel(self) -> Iterator[socket.socket]: + if self._receiver is None: + raise RuntimeError("Server log receiver is closed") # noqa: TRY003 + with self._receiver.channel() as channel: + yield channel + + @override + def prepare(self, record: logging.LogRecord) -> logging.LogRecord: + return logging.LogRecord(record.name, record.levelno, "", 0, self.format(record), (), None) + + def _receive(self, message: str) -> None: + record = logging.LogRecord("powercontext.worker", logging.INFO, "", 0, message, (), None) + try: + self._output_queue.put_nowait(record) + except Full: + self.handleError(record) + + @override + def close(self) -> None: + if self._receiver is not None: + if not self._receiver.close(): + return + self._receiver = None + if self._listener is not None: + # Drain Worker streams before placing the writer's final sentinel. + self._listener.stop() + self._listener = None + try: + self._resources.close() + except Exception: + self._notify_failure() + finally: + atexit.unregister(self.close) + super().close() class _UvicornDisplayNameFilter(logging.Filter): @@ -252,7 +395,9 @@ def filter(self, record: logging.LogRecord) -> bool: return True -def configure_server_logging(config: ServerLoggingConfig, *, role: str = "server") -> None: +def configure_server_logging( + config: ServerLoggingConfig, *, role: str = "server", channel: socket.socket | None = None +) -> None: """Configure process logging for the foreground Server command.""" formatter: dict[str, Any] @@ -268,6 +413,17 @@ def configure_server_logging(config: ServerLoggingConfig, *, role: str = "server "datefmt": "%Y-%m-%dT%H:%M:%S%z", } + if role not in {"server", "all", "api", "background", "worker"}: + raise ValueError("Unsupported logging process role") # noqa: TRY003 + if config.file and role == "worker" and channel is None: + raise ValueError("File logging workers require a Server-owned log receiver") # noqa: TRY003 + handler: dict[str, Any] = ( + {"()": _OperationalQueueHandler, "config": config, "role": role} + if config.file + else {"()": _OperationalStreamHandler, "stream": "ext://sys.stdout"} + ) + if channel is not None: + handler = {"()": _OperationalSocketHandler, "channel": channel} logging.config.dictConfig({ "version": 1, "disable_existing_loggers": False, @@ -278,8 +434,7 @@ def configure_server_logging(config: ServerLoggingConfig, *, role: str = "server "formatters": {"server": formatter}, "handlers": { "server": { - "()": _OperationalStreamHandler, - "stream": "ext://sys.stdout", + **handler, "formatter": "server", "filters": ["operational"], } @@ -305,9 +460,58 @@ class WorkerLogging: """Picklable Server-owned bootstrap; unconfigured SDK workers do not use it.""" config: ServerLoggingConfig + channel: socket.socket | None = None def __call__(self) -> None: - configure_server_logging(self.config, role="worker") + configure_server_logging(self.config, role="worker", channel=self.channel) + + +@dataclass(frozen=True) +class _WorkerLoggingPreparation: + """Keep the Server sink local and prepare a dedicated channel for each spawn.""" + + config: ServerLoggingConfig + sink: _OperationalQueueHandler + + @contextmanager + def prepare(self) -> Iterator[WorkerLogging]: + with ExitStack() as resources: + try: + channel = resources.enter_context(self.sink.worker_channel()) + except Exception: + # Channel admission cannot change the task's business result. + self.sink._notify_failure() + channel = None + yield WorkerLogging(self.config, channel) + + +@contextmanager +def worker_logging(config: ServerLoggingConfig, *, role: str) -> Iterator[WorkerLogging | _WorkerLoggingPreparation]: + """Reuse the process sink, or own a worker-only sink in an embedded Server.""" + + if not config.file: + yield WorkerLogging(config) + return + for handler in logging.getLogger("powercontext").handlers: + if isinstance(handler, _OperationalQueueHandler) and handler.config == config and handler._listener is not None: + yield _WorkerLoggingPreparation(config, handler) + return + handler = _OperationalQueueHandler(config, role=role) + try: + yield _WorkerLoggingPreparation(config, handler) + finally: + handler.close() + + +def shutdown_server_logging() -> None: + """Drain Server-owned file output after the Runtime and tracing shut down.""" + + for name in ("powercontext", "uvicorn.error"): + logger = logging.getLogger(name) + for handler in logger.handlers[:]: + if isinstance(handler, _OperationalQueueHandler): + handler.close() + logger.removeHandler(handler) __all__ = ["JsonFormatter", "OperationalContextFilter", "configure_server_logging"] diff --git a/src/powercontext/server/settings.py b/src/powercontext/server/settings.py index 962647be5b..05da9b6241 100644 --- a/src/powercontext/server/settings.py +++ b/src/powercontext/server/settings.py @@ -37,7 +37,7 @@ InferenceConfig, RuntimeConfig, ) -from powercontext.paths import default_database_path, default_seekdb_path, sqlite_url +from powercontext.paths import default_database_path, default_seekdb_path, powercontext_data_dir, sqlite_url from powercontext.transport import is_loopback_host _UNSAFE_BIND_MESSAGE = ( @@ -165,6 +165,16 @@ class ServerLoggingConfig(BaseModel): level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO" format: Literal["console", "json"] = "console" access: bool = True + file: bool = False + directory: Path = Field(default_factory=lambda: powercontext_data_dir() / "logs") + max_bytes: int = Field(default=20 * 1024 * 1024, gt=0) + backup_count: int = Field(default=10, ge=1) + compression: Literal["none", "gzip"] = "gzip" + + @field_validator("directory") + @classmethod + def resolve_directory(cls, value: Path) -> Path: + return value.expanduser().resolve() @field_validator("level", mode="before") @classmethod diff --git a/src/powercontext/service/launcher.py b/src/powercontext/service/launcher.py index af77ea3fc5..dcf64a5083 100644 --- a/src/powercontext/service/launcher.py +++ b/src/powercontext/service/launcher.py @@ -26,6 +26,7 @@ from urllib.parse import urlsplit from powercontext.server.configuration import ServerConfigurationError, server_settings_context +from powercontext.server.settings import ServerLoggingConfig from powercontext.service.environment import ProtectedEnvironmentFileError, load_protected_environment_file from powercontext.service.model import EnvironmentFileIdentity, ProbeState from powercontext.service.probe import probe_server @@ -89,6 +90,7 @@ def main(arguments: list[str] | None = None) -> int: # Help and preflight exits do not need the Server runtime dependencies. from powercontext.server import cli as server_cli + settings.logging = _service_logging(settings.logging, redirected=options.stdout is not None) server_cli._run_configured_server(settings) return 0 except (ProtectedEnvironmentFileError, ServerConfigurationError) as error: @@ -104,6 +106,14 @@ def main(arguments: list[str] | None = None) -> int: return 1 +def _service_logging(config: ServerLoggingConfig, *, redirected: bool) -> ServerLoggingConfig: + """Give file-backed native services bounded logs without duplicating stdout.""" + + if not redirected and sys.platform != "darwin": + return config + return config if "file" in config.model_fields_set else config.model_copy(update={"file": True}) + + def _environment_identity(options: argparse.Namespace) -> EnvironmentFileIdentity: fields = { "device": options.env_file_device, diff --git a/tests/e2e/test_worker_logging.py b/tests/e2e/test_worker_logging.py index ffa412ea73..9e098573c4 100644 --- a/tests/e2e/test_worker_logging.py +++ b/tests/e2e/test_worker_logging.py @@ -20,8 +20,11 @@ import io import json import logging +import multiprocessing +import subprocess import sys import time +from collections.abc import Callable from dataclasses import dataclass from typing import Any @@ -31,6 +34,7 @@ from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from opentelemetry.trace import StatusCode +from prometheus_client import CollectorRegistry from powercontext.builtin.code.telemetry import stage from powercontext.builtin.inference import ( @@ -55,15 +59,263 @@ ArtifactProcessingWorkAssignment, ArtifactProcessingWorkerCompletion, ArtifactProcessingWorkerFailure, + WorkerInitializerPreparation, ) from powercontext.server.context import bind_request_id, current_request_id, reset_request_id from powercontext.server.factory import create_server_app -from powercontext.server.logging import JsonFormatter, OperationalContextFilter, WorkerLogging +from powercontext.server.logging import JsonFormatter, OperationalContextFilter, WorkerLogging, worker_logging +from powercontext.server.operational_metrics import OperationalMetricsCollector from powercontext.server.settings import McpConfig, ServerLoggingConfig, ServerSettings from powercontext.server.telemetry import service_resource from powercontext.server.tracing import ServerTracing _PRIVATE = "private worker source/query/credential sentinel" +_MULTILINE_MESSAGE = "child record\ncontinued café" + + +def _file_logging_probe(initializer: WorkerLogging) -> None: + initializer() + logging.getLogger("powercontext.worker.probe").info(_MULTILINE_MESSAGE, extra={"source_body": lambda: None}) + + +def _waiting_logging_worker(assignment: ArtifactProcessingWorkAssignment) -> ArtifactProcessingWorkerCompletion: + logging.getLogger("powercontext.worker.probe").info("Ready for termination", extra={"event": "probe.ready"}) + time.sleep(60) + return ArtifactProcessingWorkerCompletion() + + +def _noisy_logging_worker(assignment: ArtifactProcessingWorkAssignment) -> ArtifactProcessingWorkerCompletion: + for index in range(2000): + logging.getLogger("powercontext.worker.probe").info( + "Worker observation %s %s", index, "x" * 2000, extra={"source_body": _PRIVATE} + ) + return ArtifactProcessingWorkerCompletion() + + +def _closing_logging_probe(initializer: WorkerLogging, *, truncate: bool) -> None: + initializer() + logging.getLogger("powercontext.worker.probe").info("Delivered observation") + if truncate: + assert initializer.channel is not None + # Leave one incomplete envelope before the process closes its socket. + initializer.channel.sendall(b'"' + _PRIVATE.encode("utf-8")) + + +@pytest.mark.parametrize("truncate", [False, True]) +def test_worker_stream_close_preserves_output_and_counts_discards(tmp_path, capfd, truncate) -> None: + registry = CollectorRegistry() + registry.register(OperationalMetricsCollector(role="background")) + metric = "powercontext_server_telemetry_log_records_dropped_total" + before = registry.get_sample_value(metric) + assert before is not None + config = ServerLoggingConfig(file=True, directory=tmp_path, format="json") + with worker_logging(config, role="background") as initializer: + assert isinstance(initializer, WorkerInitializerPreparation) + with initializer.prepare() as child_initializer: + assert isinstance(child_initializer, WorkerLogging) + process = multiprocessing.get_context("spawn").Process( + target=_closing_logging_probe, args=(child_initializer,), kwargs={"truncate": truncate} + ) + process.start() + try: + process.join(timeout=30) + assert process.exitcode == 0 + finally: + if process.is_alive(): + process.kill() + process.join(timeout=5) + process.close() + assert registry.get_sample_value(metric) == before + int(truncate) + output = capfd.readouterr() + assert ("observability.log_failed" in output.err) == truncate + records = (tmp_path / "background.log").read_text(encoding="utf-8") + assert [record["message"] for record in _json_records(records)] == ["Delivered observation"] + assert output.out == "" + assert _PRIVATE not in records + output.err + + +def test_worker_completion_does_not_wait_for_slow_log_consumer(tmp_path) -> None: + script = """ +import asyncio +import logging +import sys +from pathlib import Path +from tests.e2e.test_worker_logging import _noisy_logging_worker +from powercontext.builtin.persistence.supervision import ArtifactProcessingFence +from powercontext.builtin.runtime.artifact_processing import SpawnArtifactProcessingWorkerLauncher +from powercontext.builtin.runtime.processing_contracts import ( + ArtifactProcessingWorkAssignment, ArtifactProcessingWorkerCompletion, +) +from powercontext.server.logging import configure_server_logging, shutdown_server_logging, worker_logging +from powercontext.server.settings import ServerLoggingConfig + +blocked = Path(sys.argv[3]) +original_emit = logging.FileHandler.emit +def slow_emit(handler, record): + if not blocked.exists(): + blocked.write_text("blocked", encoding="utf-8") + sys.stdin.read(1) + original_emit(handler, record) +logging.FileHandler.emit = slow_emit + +config = ServerLoggingConfig(file=True, directory=Path(sys.argv[1]), format="json") +configure_server_logging(config, role="all") +async def run(initializer): + assignment = ArtifactProcessingWorkAssignment( + binding_name="memory-source-window", scope_id="success", artifact_family="memory", + claimed_request_generation=1, worker_id="noisy-attempt", + fence=ArtifactProcessingFence( + supervisor_group="global", holder_id="owner", supervisor_generation=1, lease_mode="single-process", + ), + ) + handle = await SpawnArtifactProcessingWorkerLauncher(_noisy_logging_worker, initializer=initializer).start(assignment) + try: + assert await asyncio.wait_for(handle.wait(), timeout=30) == ArtifactProcessingWorkerCompletion() + finally: + await handle.terminate() +with worker_logging(config, role="all") as initializer: + asyncio.run(run(initializer)) +Path(sys.argv[2]).write_text("completed", encoding="utf-8") +shutdown_server_logging() +""" + completed = tmp_path / "completed" + blocked = tmp_path / "blocked" + log_directory = tmp_path / "logs" + process = subprocess.Popen( + [sys.executable, "-c", script, str(log_directory), str(completed), str(blocked)], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + deadline = time.monotonic() + 45 + # Hold file writes until the business result is available, then release the sink. + while not completed.exists() and process.poll() is None and time.monotonic() < deadline: + time.sleep(0.01) + finished_without_draining = completed.exists() and blocked.exists() + stdout, stderr = process.communicate(input="\n", timeout=30) + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=10) + assert process.returncode == 0, stderr + assert finished_without_draining + assert completed.read_text(encoding="utf-8") == "completed" + output = (log_directory / "all.log").read_text(encoding="utf-8") + assert "Worker observation" in output + assert stdout == "" + assert _PRIVATE not in output + stderr + + +def test_terminated_worker_does_not_interrupt_shared_log_output(tmp_path, capfd) -> None: + config = ServerLoggingConfig(file=True, directory=tmp_path, format="json") + + async def run(initializer: Callable[[], None] | WorkerInitializerPreparation) -> None: + assignment = ArtifactProcessingWorkAssignment( + binding_name="memory-source-window", + scope_id="success", + artifact_family="memory", + claimed_request_generation=1, + fence=ArtifactProcessingFence( + supervisor_group="global", holder_id="owner", supervisor_generation=1, lease_mode="single-process" + ), + worker_id="terminated-attempt", + ) + handle = await SpawnArtifactProcessingWorkerLauncher(_waiting_logging_worker, initializer=initializer).start( + assignment + ) + try: + async with asyncio.timeout(30): + while True: + output = (tmp_path / "background.log").read_text(encoding="utf-8") + if any(record.get("event") == "probe.ready" for record in _json_records(output)): + break + await asyncio.sleep(0.01) + finally: + await handle.terminate() + surviving = await SpawnArtifactProcessingWorkerLauncher(_logging_worker, initializer=initializer).start( + ArtifactProcessingWorkAssignment( + binding_name=assignment.binding_name, + scope_id="success", + artifact_family="memory", + claimed_request_generation=1, + fence=assignment.fence, + worker_id="surviving-attempt", + ) + ) + try: + assert await asyncio.wait_for(surviving.wait(), timeout=30) == ArtifactProcessingWorkerCompletion() + finally: + await surviving.terminate() + + with worker_logging(config, role="background") as initializer: + asyncio.run(run(initializer)) + output = capfd.readouterr() + assert output.out == "" and output.err == "" + records = _json_records((tmp_path / "background.log").read_text(encoding="utf-8")) + assert any( + record.get("event") == "probe.ready" and record["worker_id"] == "terminated-attempt" for record in records + ) + assert any( + record.get("event") == "artifact_processing.worker.returned" and record["worker_id"] == "surviving-attempt" + for record in records + ) + + +@pytest.mark.parametrize("log_format", ["json", "console"]) +def test_configured_file_sink_remains_open_after_spawned_worker_cleanup(tmp_path, log_format) -> None: + script = """ +import logging +import multiprocessing +import sys +from pathlib import Path +from tests.e2e.test_worker_logging import _file_logging_probe +from powercontext.server.logging import configure_server_logging, shutdown_server_logging, worker_logging +from powercontext.server.settings import ServerLoggingConfig + +config = ServerLoggingConfig(file=True, directory=Path(sys.argv[1]), format=sys.argv[2]) +configure_server_logging(config, role="all") +logging.getLogger("powercontext.parent").info("parent before") +with worker_logging(config, role="all") as initializer: + with initializer.prepare() as child_initializer: + process = multiprocessing.get_context("spawn").Process(target=_file_logging_probe, args=(child_initializer,)) + process.start() + try: + process.join(timeout=30) + assert process.exitcode == 0 + finally: + if process.is_alive(): + process.kill() + process.join(timeout=5) + process.close() +logging.getLogger("powercontext.parent").info("parent after") +shutdown_server_logging() +""" + result = subprocess.run( + [sys.executable, "-c", script, str(tmp_path), log_format], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + assert result.stdout == "" and result.stderr == "" + output = (tmp_path / "all.log").read_text(encoding="utf-8") + assert "source_body" not in output + if log_format == "console": + assert _MULTILINE_MESSAGE in output + assert output.count("parent before") == output.count("parent after") == 1 + assert 'service_role="worker"' in output and 'service_role="all"' in output + return + records = _json_records(output) + assert len(records) == 3 + by_message = {record["message"]: record for record in records} + assert set(by_message) == {"parent before", _MULTILINE_MESSAGE, "parent after"} + assert by_message["parent before"]["service_instance_id"] == by_message["parent after"]["service_instance_id"] + assert by_message["parent before"]["service_instance_id"] != by_message[_MULTILINE_MESSAGE]["service_instance_id"] + assert by_message["parent before"]["service_role"] == by_message["parent after"]["service_role"] == "all" + assert by_message[_MULTILINE_MESSAGE]["service_role"] == "worker" + assert all("source_body" not in record for record in records) class _WorkerCancelled(asyncio.CancelledError): @@ -464,7 +716,10 @@ async def run() -> None: assert output.out == "" and output.err == "" -def test_server_composition_initializes_logging_in_the_real_family_worker(capfd, tmp_path, monkeypatch) -> None: +@pytest.mark.parametrize("file_logging", [False, True]) +def test_server_composition_initializes_logging_in_the_real_family_worker( + capfd, tmp_path, monkeypatch, file_logging +) -> None: parent_logger = logging.getLogger("powercontext") handler = logging.StreamHandler(sys.stdout) handler.setFormatter(JsonFormatter()) @@ -477,7 +732,7 @@ def test_server_composition_initializes_logging_in_the_real_family_worker(capfd, database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'worker-logging.db'}"), inference=InferenceConfig(generation_model="test"), runtime=RuntimeConfig(artifact_processing_families=("memory",), memory_schedule_seconds=0.05), - logging=ServerLoggingConfig(format="json"), + logging=ServerLoggingConfig(format="json", file=file_logging, directory=tmp_path / "logs"), mcp=McpConfig(enabled=False), ) ) @@ -498,7 +753,10 @@ def test_server_composition_initializes_logging_in_the_real_family_worker(capfd, break time.sleep(0.05) records.extend(_json_records(capfd.readouterr().out)) - child = [record for record in records if record.get("event") == "artifact_processing.worker.started"] + worker_records = ( + _json_records((tmp_path / "logs" / "all.log").read_text(encoding="utf-8")) if file_logging else records + ) + child = [record for record in worker_records if record.get("event") == "artifact_processing.worker.started"] assert child, "Server must configure logging in its actual built-in child, not a substituted entrypoint" assert child[0]["service_role"] == "worker" and child[0]["family"] == "memory" assert "request_id" not in child[0] @@ -511,4 +769,54 @@ def test_server_composition_initializes_logging_in_the_real_family_worker(capfd, assert controller[0]["worker_id"] == child[0]["worker_id"] assert controller[0]["service_role"] == "all" assert controller[0]["service_instance_id"] != child[0]["service_instance_id"] - assert _PRIVATE not in json.dumps(records) + assert _PRIVATE not in json.dumps([*records, *worker_records]) + + +@pytest.mark.parametrize("log_format", ["json", "console"]) +def test_concurrent_spawned_workers_share_file_output(tmp_path, capfd, log_format) -> None: + config = ServerLoggingConfig(file=True, directory=tmp_path, format=log_format) + + async def run(initializer: Callable[[], None] | WorkerInitializerPreparation) -> None: + launcher = SpawnArtifactProcessingWorkerLauncher(_logging_worker, initializer=initializer) + assignments = [ + ArtifactProcessingWorkAssignment( + binding_name="memory-source-window", + scope_id="success", + artifact_family="memory", + claimed_request_generation=1, + fence=ArtifactProcessingFence( + supervisor_group="global", holder_id="owner", supervisor_generation=1, lease_mode="single-process" + ), + worker_id=f"file-attempt-{index}", + ) + for index in range(3) + ] + handles = await asyncio.gather(*(launcher.start(assignment) for assignment in assignments)) + try: + results = await asyncio.wait_for(asyncio.gather(*(handle.wait() for handle in handles)), timeout=30) + assert results == [ArtifactProcessingWorkerCompletion()] * 3 + finally: + await asyncio.gather(*(handle.terminate() for handle in handles)) + + with worker_logging(config, role="background") as initializer: + asyncio.run(run(initializer)) + output = (tmp_path / "background.log").read_text(encoding="utf-8") + assert _PRIVATE not in output + captured = capfd.readouterr() + assert captured.out == "" and captured.err == "" + if log_format == "json": + records = _json_records(output) + assert len(records) == 9 + assert len({record["service_instance_id"] for record in records}) == 3 + for index in range(3): + attempt = [record for record in records if record["worker_id"] == f"file-attempt-{index}"] + assert {record["event"] for record in attempt} == { + "artifact_processing.worker.started", + "artifact_processing.domain", + "artifact_processing.worker.returned", + } + assert all(record["service_role"] == "worker" and "request_id" not in record for record in records) + else: + assert len(output.splitlines()) == 9 + for index in range(3): + assert output.count(f'worker_id="file-attempt-{index}"') == 3 diff --git a/tests/test_server_logging.py b/tests/test_server_logging.py index 6dc080dc40..38d8a08a67 100644 --- a/tests/test_server_logging.py +++ b/tests/test_server_logging.py @@ -14,10 +14,12 @@ from __future__ import annotations +import gzip import json import logging import subprocess import sys +from pathlib import Path import pytest from fastapi.testclient import TestClient @@ -28,6 +30,193 @@ from powercontext.server.settings import McpConfig, ServerLoggingConfig, ServerSettings +@pytest.mark.parametrize("compression", ["none", "gzip"]) +@pytest.mark.parametrize("log_format", ["console", "json"]) +def test_file_logging_rotates_and_limits_readable_archives(tmp_path: Path, compression: str, log_format: str) -> None: + script = """ +import logging +import sys +from pathlib import Path +from powercontext.server.logging import configure_server_logging, shutdown_server_logging +from powercontext.server.settings import ServerLoggingConfig + +config = ServerLoggingConfig( + file=True, directory=Path(sys.argv[1]), format=sys.argv[3], + max_bytes=2048, backup_count=2, compression=sys.argv[2], +) +configure_server_logging(config) +logger = logging.getLogger("powercontext.file") +for index in range(100): + logger.info("record %s", index, extra={"source_body": lambda: None}) +logger.info("private-message-sentinel", extra={"operation": object()}) +logger.info("final record", extra={"event": "probe.completed", "outcome": "success"}) +shutdown_server_logging() +configure_server_logging(config) +logging.getLogger("uvicorn.error").info("reopened record") +shutdown_server_logging() +""" + directory = tmp_path / "nested" / "logs" + result = subprocess.run( + [sys.executable, "-c", script, str(directory), compression, log_format], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + archives = sorted(directory.glob("server.log.[0-9]*")) + assert len(archives) == 2 + counts = {"final record": 0, "reopened record": 0} + for path in [directory / "server.log", *archives]: + assert (path.suffix == ".gz") == (compression == "gzip" and path in archives) + open_log = gzip.open if path.suffix == ".gz" else open + with open_log(path, "rt", encoding="utf-8") as stream: + for line in stream: + assert "private-message-sentinel" not in line and "source_body" not in line + assert "record 0 " not in line + for message in counts: + counts[message] += line.count(message) + if log_format == "json": + record = json.loads(line) + if record["message"] == "reopened record": + assert record["logger"] == "uvicorn" + assert counts == {"final record": 1, "reopened record": 1} + assert result.stderr.splitlines() == ['{"event":"observability.log_failed","outcome":"failure"}'] + assert result.stdout == "" + + +def test_file_logging_flushes_on_normal_process_exit(tmp_path: Path) -> None: + script = """ +import logging +import sys +from pathlib import Path +from powercontext.server.logging import configure_server_logging +from powercontext.server.settings import ServerLoggingConfig +configure_server_logging(ServerLoggingConfig(file=True, directory=Path(sys.argv[1]), format="json")) +logging.getLogger("powercontext.file").info("last record") +""" + result = subprocess.run( + [sys.executable, "-c", script, str(tmp_path)], check=True, capture_output=True, text=True, timeout=60 + ) + assert result.stderr == "" + assert result.stdout == "" + assert json.loads((tmp_path / "server.log").read_text(encoding="utf-8"))["message"] == "last record" + + +@pytest.mark.skipif(not Path("/dev/full").exists(), reason="Requires the OS disk-full device") +def test_full_disk_does_not_dump_records_or_interrupt_processing(tmp_path: Path) -> None: + (tmp_path / "server.log").symlink_to("/dev/full") + script = """ +import logging +import sys +from pathlib import Path +from powercontext.server.logging import configure_server_logging, shutdown_server_logging +from powercontext.server.settings import ServerLoggingConfig + +configure_server_logging(ServerLoggingConfig( + file=True, directory=Path(sys.argv[1]), format="json", +)) +for index in range(10): + logging.getLogger("powercontext.file").info("private-message-sentinel %s", index) +shutdown_server_logging() +print("processing continues") +""" + result = subprocess.run( + [sys.executable, "-c", script, str(tmp_path)], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + assert result.stdout == "processing continues\n" + assert set(result.stderr.splitlines()) == {'{"event":"observability.log_failed","outcome":"failure"}'} + + +def test_file_logging_rejects_a_second_writer_until_the_owner_closes(tmp_path: Path) -> None: + script = """ +import logging +import subprocess +import sys +from pathlib import Path +from powercontext.server.logging import configure_server_logging, shutdown_server_logging +from powercontext.server.settings import ServerLoggingConfig + +child = ''' +import logging +import sys +from pathlib import Path +from powercontext.server.logging import configure_server_logging, shutdown_server_logging +from powercontext.server.settings import ServerLoggingConfig +try: + configure_server_logging(ServerLoggingConfig(file=True, directory=Path(sys.argv[1]), format="json")) +except ValueError: + sys.exit(7) +logging.getLogger("powercontext.file").info("second writer") +shutdown_server_logging() +''' +configure_server_logging(ServerLoggingConfig(file=True, directory=Path(sys.argv[1]), format="json")) +logging.getLogger("powercontext.file").info("first writer") +result = subprocess.run([sys.executable, "-c", child, sys.argv[1]], capture_output=True, text=True, timeout=30) +assert result.returncode == 7 +assert result.stdout == result.stderr == "" +shutdown_server_logging() +subprocess.run([sys.executable, "-c", child, sys.argv[1]], check=True, timeout=30) +""" + result = subprocess.run( + [sys.executable, "-c", script, str(tmp_path)], check=True, capture_output=True, text=True, timeout=90 + ) + assert result.stdout == result.stderr == "" + records = [json.loads(line) for line in (tmp_path / "server.log").read_text(encoding="utf-8").splitlines()] + assert [record["message"] for record in records] == ["first writer", "second writer"] + + +def test_file_logging_does_not_start_auxiliary_processes(tmp_path) -> None: + script = """ +import logging +import multiprocessing +import sys +from pathlib import Path +from powercontext.server.logging import configure_server_logging, shutdown_server_logging +from powercontext.server.settings import ServerLoggingConfig + +config = ServerLoggingConfig(file=True, directory=Path(sys.argv[1]), format="json") +configure_server_logging(config) +assert multiprocessing.active_children() == [] +logging.getLogger("powercontext.file").info("first record") +shutdown_server_logging() +configure_server_logging(config) +logging.getLogger("powercontext.file").info("reopened record") +shutdown_server_logging() +assert multiprocessing.active_children() == [] +""" + result = subprocess.run( + [sys.executable, "-c", script, str(tmp_path)], check=True, capture_output=True, text=True, timeout=60 + ) + assert result.stdout == "" and result.stderr == "" + records = [json.loads(line) for line in (tmp_path / "server.log").read_text(encoding="utf-8").splitlines()] + assert [record["message"] for record in records] == ["first record", "reopened record"] + + +def test_logging_file_settings_use_environment_and_user_data_directory(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("POWERCONTEXT_HOME", str(tmp_path)) + monkeypatch.setenv("POWERCONTEXT_SERVER_LOGGING_FILE", "true") + monkeypatch.setenv("POWERCONTEXT_SERVER_LOGGING_DIRECTORY", str(tmp_path / "custom-logs")) + assert ServerSettings().logging.directory == tmp_path / "custom-logs" + monkeypatch.delenv("POWERCONTEXT_SERVER_LOGGING_DIRECTORY") + monkeypatch.setenv("POWERCONTEXT_SERVER_LOGGING_MAX_BYTES", "4096") + monkeypatch.setenv("POWERCONTEXT_SERVER_LOGGING_BACKUP_COUNT", "3") + monkeypatch.setenv("POWERCONTEXT_SERVER_LOGGING_COMPRESSION", "none") + config = ServerSettings().logging + assert config.file + assert config.directory == tmp_path / "logs" + assert config.max_bytes == 4096 and config.backup_count == 3 and config.compression == "none" + + +@pytest.mark.parametrize("values", [{"max_bytes": 0}, {"backup_count": 0}, {"compression": "zip"}]) +def test_logging_rejects_unbounded_rotation_and_unsupported_compression(values) -> None: + with pytest.raises(ValueError): + ServerLoggingConfig(**values) + + def test_operational_filter_uses_bound_request_id_without_overwriting_an_explicit_id() -> None: from powercontext.server.context import bind_request_id, reset_request_id diff --git a/tests/test_service.py b/tests/test_service.py index ece742aaf1..ad99cbf753 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1933,6 +1933,35 @@ def run_server(_settings: object) -> None: assert stderr_path.read_text(encoding="utf-8") == "server error\n" +@pytest.mark.parametrize("file_enabled", [True, False]) +def test_service_launcher_uses_bounded_file_logging_for_redirected_output(tmp_path, monkeypatch, file_enabled) -> None: + observed = [] + environment = tmp_path / "service.env" + values = "POWERCONTEXT_SERVER_LOGGING_FORMAT=json\n" if file_enabled else "POWERCONTEXT_SERVER_LOGGING_FILE=false\n" + environment.write_text(values, encoding="utf-8") + environment.chmod(0o600) + _secure_windows_file(environment) + definition = _definition( + tmp_path, data_dir=str(tmp_path), env_file=load_protected_environment_file(environment).identity + ) + monkeypatch.setattr( + service_launcher, "probe_server", lambda _endpoint: ProbeResult(ProbeState.UNREACHABLE, "not listening") + ) + monkeypatch.setattr( + "powercontext.server.cli._run_configured_server", lambda settings: observed.append(settings.logging) + ) + assert ( + service_launcher.main([ + *definition.launcher_arguments()[3:], + "--stdout", + str(tmp_path / "logs" / "server.stdout.log"), + ]) + == 0 + ) + assert observed[0].file == file_enabled + assert observed[0].directory == tmp_path / "logs" + + @pytest.mark.skipif(os.name == "nt", reason="launchd bootstrap retry state is POSIX-specific") def test_launchd_launcher_stops_after_a_bounded_number_of_rapid_failures( tmp_path: Path,