mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
feat(cli): Add bounded daemon log rotation (#6969)
* feat(cli): add bounded daemon log rotation Co-authored-by: Qwen-Coder <[email protected]> * codex: address PR review feedback (#6969) Co-authored-by: Qwen-Coder <[email protected]> * codex: address PR review feedback (#6969) Co-authored-by: Qwen-Coder <[email protected]> * codex: address PR review feedback (#6969) Co-authored-by: Qwen-Coder <[email protected]> * codex: address PR review feedback (#6969) Co-authored-by: Qwen-Coder <[email protected]> * codex: address PR review feedback (#6969) Co-authored-by: Qwen-Coder <[email protected]> --------- Co-authored-by: Qwen-Coder <[email protected]>
This commit is contained in:
jinyeandQwen-Coder authored and GitHub committed
1 parent
1ec80b10d1
commit
67e581aeba
18 files changed
+3867
-245
No files matched your search
@@ -0,0 +1,160 @@
|
||||
# Stable and Bounded Daemon Logs
|
||||
|
||||
- **Status:** Implemented
|
||||
- **Date:** 2026-07-15
|
||||
- **Scope:** `qwen serve` file logging, lifecycle ownership, access-log admission, daemon status, and the TypeScript SDK status mirror
|
||||
|
||||
## Decision
|
||||
|
||||
Each runtime log namespace has one stable active path:
|
||||
|
||||
```text
|
||||
${runtimeBaseDir}/debug/daemon/daemon.log
|
||||
```
|
||||
|
||||
Normal restarts append to that path. The fixed policy is:
|
||||
|
||||
| Limit | Value |
|
||||
| ------------------------------------- | -----------: |
|
||||
| Active file | 10 MiB |
|
||||
| Archives per family | 4 |
|
||||
| Rendered file record | 256 KiB |
|
||||
| Accepted but unsettled file payload | 4 MiB |
|
||||
| Stable lease stale/update | 60 s / 10 s |
|
||||
| Stable/maintenance acquisition budget | 1 s / 250 ms |
|
||||
| Public logger close budget | 2 s |
|
||||
|
||||
These values are intentionally not CLI flags, environment variables, or settings. A healthy stable family occupies at most about 50 MiB. Retaining the most recent inactive fallback family brings the converged namespace to about 100 MiB. Live or not-yet-stale fallback owners are never deleted, so temporary use can grow with the number of possibly live daemons.
|
||||
|
||||
Every start generates a random 128-bit `runId`. Every file record begins with immutable `runId` and daemon PID context. Caller context cannot replace those values. Stderr keeps the existing formatting and field order.
|
||||
|
||||
## Namespace and ownership
|
||||
|
||||
The configured log directory is the ownership and retention namespace. Workspace, listener port, and PID are not storage identities: a daemon can host several workspaces, port zero is dynamic, ports can advance on conflict, and embedded daemons can share a PID.
|
||||
|
||||
The stable family is owned by a lifetime `proper-lockfile` lease. A contender that cannot acquire it writes to:
|
||||
|
||||
```text
|
||||
debug/daemon/runs/run-<32-hex-runId>/daemon.log
|
||||
```
|
||||
|
||||
It holds that family's `.owner.lock` for its lifetime and never promotes into the stable family during the run. The boot banner and full daemon status are authoritative for the selected path. `runs/recent-fallback` is only a validated discovery hint.
|
||||
|
||||
Fallback allocation and cleanup are serialized by `runs/.maintenance.lock`. Cleanup retains every busy owner family and at most one inactive family. It prefers a valid locator, then the newest active-log mtime, then basename as a deterministic tie-breaker. A non-lock cleanup error or failed deletion rejects the allocation so a damaged namespace does not accumulate a new directory on every start.
|
||||
|
||||
Clean fallback close acquires maintenance ownership, releases its owner lease, retains the current family, removes other inactive families, and repairs the locator. If maintenance ownership is unavailable, close releases only the owner lease and leaves repair to a later start.
|
||||
|
||||
## Filesystem layout
|
||||
|
||||
```text
|
||||
debug/daemon/
|
||||
├── daemon.log
|
||||
├── latest -> daemon.log
|
||||
├── .stable-writer.lock/
|
||||
├── archive/
|
||||
│ └── daemon-000000000001-20260715T031415926Z-a1b2c3d4.log
|
||||
└── runs/
|
||||
├── .maintenance.lock/
|
||||
├── recent-fallback
|
||||
└── run-6a45c211000000000000000000000000/
|
||||
├── .owner.lock/
|
||||
├── daemon.log
|
||||
└── archive/
|
||||
```
|
||||
|
||||
Only strictly matching regular archive files participate in retention. Legacy `serve-<pid>.log` and `serve-<pid>-<workspaceHash>.log` files are neither migrated nor deleted.
|
||||
|
||||
New directories use mode `0700`; new active logs and locator temporary files use `0600`. Existing object permissions are not rewritten. `latest` is updated only by a successful stable owner and remains best-effort where symlinks are unavailable.
|
||||
|
||||
## File records and queue
|
||||
|
||||
File records are truncated on a valid UTF-8 boundary. The final record, including an original-byte-count marker and newline, is at most 256 KiB. Its stderr copy is not truncated.
|
||||
|
||||
One Promise queue preserves file mutation order. Accepted but unsettled record bytes are accounted synchronously. A record that would raise the queue above 4 MiB loses only its file copy; the logger increments `droppedRecords` and `droppedBytes` and warns once for that overflow episode.
|
||||
|
||||
After capacity recovers, the next caller record is preceded by a file-only warning named `daemon file log records dropped`. It reports the unreported record and byte totals and does not recursively contribute to them. Close makes one final attempt after draining the queue.
|
||||
|
||||
Each queue task catches its own failure and releases its pending-byte accounting in `finally`; the shared tail never remains rejected. If an active append rejects, its result is unknown: the logger records `write_failed`, stops all subsequent file mutation for that run, and does not claim the failed record as an exact loss. Later records that are deliberately skipped are counted.
|
||||
|
||||
Lease compromise likewise stops new file mutations immediately. A single filesystem operation that already started may finish, but no later append, rotation, or deletion starts through that family.
|
||||
|
||||
## Rotation transaction
|
||||
|
||||
Before a record would make the active file exceed 10 MiB, the logger:
|
||||
|
||||
1. verifies that `archive/` is a real, non-symlink directory;
|
||||
2. removes the oldest generated archives until at most three remain;
|
||||
3. chooses a nonexistent name containing a 12-digit generation, UTC timestamp, and random suffix;
|
||||
4. atomically renames the active path to that archive name;
|
||||
5. appends the triggering record to a new `daemon.log` with mode `0600`; and
|
||||
6. commits the in-memory size and generation state.
|
||||
|
||||
Thus a family produced by this implementation has at most one active file and four archives. If the new-active append fails, the previous active file remains complete in the newest archive.
|
||||
|
||||
Archive validation, pruning, naming, or rename failure drops the record rather than allowing the active file to cross 10 MiB. Rotation is retried at most once per 60 seconds while smaller records that still fit may continue. There is no special ENOSPC/EDQUOT deletion-and-retry protocol and no rejected-append truncate rollback because neither can prove the file's resulting state.
|
||||
|
||||
Initialization reads the active file's real size. If its last byte is not a newline and the boot record does not first rotate it, the logger inserts a newline and marks the boot record with `previousTailIncomplete=true`. If the stable boot probe cannot safely write, it releases the stable lease and attempts one fallback family. A failed fallback probe yields degraded stderr-only logging.
|
||||
|
||||
## Logger state and lifecycle
|
||||
|
||||
```ts
|
||||
type DaemonLogMode = 'stable' | 'fallback' | 'stderr-only';
|
||||
type DaemonLogHealth = 'ok' | 'degraded';
|
||||
type DaemonLogIssue =
|
||||
| 'init_failed'
|
||||
| 'rotation_failed'
|
||||
| 'retention_failed'
|
||||
| 'queue_overflow'
|
||||
| 'write_failed'
|
||||
| 'lease_compromised';
|
||||
```
|
||||
|
||||
`getStatus()` returns the run identity, mode, health, ordered issues, and loss counters. `QWEN_DAEMON_LOG_FILE=0|false|off|no` returns a healthy stderr-only logger without accessing the filesystem: `info`, `warn`, and `error` still write stderr, while `raw` remains file-only and therefore does nothing.
|
||||
|
||||
`close()` is idempotent and non-rejecting. It synchronously stops accepting file copies, while structured stderr calls remain usable. Its background finalizer drains the queue, attempts the final loss summary, performs fallback cleanup, and releases the lifetime lease. The public Promise waits at most two seconds; a timeout does not release the lease early, and the finalizer remains alive until started I/O settles. `flush()` keeps its unbounded queue-snapshot semantics. Forced signal paths and retryable resource-close failures race it against 250 ms.
|
||||
|
||||
Logger ownership moves through:
|
||||
|
||||
```text
|
||||
startup -> published handle -> terminal close
|
||||
\-> startup signal -> terminal close
|
||||
```
|
||||
|
||||
An internal close before handle publication drains daemon resources without waiting for the logger queue, then leaves the logger to the outer startup-error owner. That owner records `daemon startup failed` and closes it. A terminal published or signal-owned close seals access logging, records `daemon stopped`, and closes the logger even when resource shutdown returns a non-retryable error; the original resource error remains the returned error. Terminal diagnostic writes are best-effort so an unavailable stderr cannot replace the original failure or skip logger cleanup. A retryable channel-worker/service-lease failure keeps the logger open, uses the bounded flush above, and does not record `daemon stopped`.
|
||||
|
||||
## Access-log admission
|
||||
|
||||
Each runtime Express app owns a constant-space token bucket with burst 60 and refill 2 records/second, measured with a monotonic clock. Clock retreat never moves the refill baseline backward. Health, heartbeat, and successful SSE exclusions are unchanged.
|
||||
|
||||
Route, session ID, and the first raw `x-qwen-client-id` occurrence are capped at 2 KiB, 256 bytes, and 256 bytes on UTF-8 boundaries. Truncated values carry an original-byte-count context field. Using the first raw header avoids merged duplicate headers becoming a new cardinality source.
|
||||
|
||||
When no token is available, only five fixed counters are retained: 2xx, 3xx, 4xx, 5xx, and other. On recovery a WARN `access logs suppressed` summary consumes the next token before any individual record. If that was the only token, the current request joins the next summary. Shutdown seals the controller after normal listener drain or the secondary deadline, emits a final summary, ignores late finish callbacks, and then records `daemon stopped`.
|
||||
|
||||
Rate limiting affects diagnostics only; it never changes the HTTP result. Suppressed individual records reach neither stderr nor file, while summaries reach both.
|
||||
|
||||
## Daemon status and SDK
|
||||
|
||||
Each status response takes one logger snapshot. Summary and full responses may contain:
|
||||
|
||||
- `daemon.runId`
|
||||
- `daemon.logMode`
|
||||
- `daemon.logHealth`
|
||||
|
||||
Full responses may additionally contain `daemon.logPath`, `daemon.logIssues`, `daemon.logDroppedRecords`, and `daemon.logDroppedBytes`. Degraded logging adds a path-free top-level `daemon_log_degraded` warning to the existing rollup. The TypeScript SDK mirrors the optional fields and closed unions. No capability tag or client upgrade is required.
|
||||
|
||||
Opt-out reports `stderr-only/ok`; ordinary stable contention reports `fallback/ok`; filesystem initialization failure reports degraded logging with `init_failed`.
|
||||
|
||||
## Operational and compatibility boundaries
|
||||
|
||||
- Use separate runtime directories for independent retention or audit namespaces.
|
||||
- On macOS/Linux use `tail -F daemon.log`; on every platform, viewers must reopen the pathname after rotation.
|
||||
- Do not configure external logrotate to mutate `daemon.log`. Copying or shipping it is safe; renaming, truncating, or deleting it breaks the in-memory size model.
|
||||
- There is no age expiry, compression, fsync durability, or absolute bound during concurrent-daemon or crash-restart storms inside the stale window.
|
||||
- Same-user tampering, false stale takeover, filesystem calls that never return, sudden power loss, and Windows readers that prevent rename are handled by safe degradation, not by platform-specific no-follow, fsync, or process-admission protocols.
|
||||
- Downgrading remains possible; older versions simply resume creating PID-named files.
|
||||
|
||||
## Verification strategy
|
||||
|
||||
Unit coverage includes formatting, immutable file context, stable reuse, UTF-8 truncation, rotation bounds, incomplete tails, queue overflow summaries, poisoned appends, active and post-release compromised leases, bounded close and retry flushes, stable/fallback concurrency, fallback retention, cleanup refusal, lifecycle diagnostic failures, access token admission, shutdown sealing, status snapshots, isolated test runtime namespaces, and SDK type surface.
|
||||
|
||||
Process-level verification uses a built bundle and isolated runtime directory for restart reuse, real-threshold rotation, stable/fallback concurrency, signal lease release, SIGKILL stale-window behavior, access aggregation, legacy-file preservation, and opt-out without filesystem access. The CI platform matrix must exercise direct active paths on macOS, Linux, and Windows; Windows additionally verifies safe degradation when a reader prevents active/archive rename.
|
||||
@@ -11,8 +11,8 @@
|
||||
| `QWEN_SERVE_DEBUG` stderr logs | `bridge.ts` and call sites | Env values `1` / `true` / `on` / `yes` (case-insensitive) print `qwen serve debug: ...` lines to stderr. |
|
||||
| OpenTelemetry span instrumentation | `server.ts` `daemonTelemetryMiddleware` | Classified daemon API requests that reach the telemetry middleware are wrapped in `withDaemonRequestSpan`; attributes include canonical route, workspace hash when resolved, sessionId, clientId, and status code. Permission routes have dedicated spans. Prompt lifecycle is traced end-to-end. Configuration lives in `settings.json` `telemetry`. |
|
||||
| OpenTelemetry daemon perf metrics | `telemetry/*event-loop-lag*`, `daemon-metrics` | Event loop lag gauges for daemon and ACP child processes, plus daemon-child pipe message byte histograms. |
|
||||
| `DaemonLogger` structured file logs | `serve/daemon-logger.ts` | Structured JSON-like log lines are written to a file. Boot prints `daemon log -> <path>`. Supports `info` / `warn` / `error` levels, with structured fields such as `route`, `sessionId`, `clientId`, `childPid`, and `channelId`. |
|
||||
| Per-request access-log middleware | `server.ts`, registered before `bearerAuth` | Logs `method`, `path`, `status`, `durationMs`, `sessionId`, and `clientId` after each request. Skips `GET /health` and heartbeat. 4xx+ uses `warn`; success uses `info`. |
|
||||
| `DaemonLogger` structured file logs | `serve/daemon-logger.ts` | Appends to a stable, size-rotated `daemon.log`. File records include `runId` and PID. Boot prints the selected stable/fallback path; full status exposes health, issues, and file-copy loss counters. |
|
||||
| Per-request access-log middleware | `server/access-log.ts` | Logs method/path, status, duration, session, and first raw client ID after each request. A 60-token burst / 2-per-second bucket aggregates excess traffic into five fixed status counters. Health, heartbeat, and successful SSE exclusions remain. |
|
||||
| `/health` | `server.ts` route | Liveness probe; `?deep=1` returns extended details. |
|
||||
| `/capabilities` | `server.ts` route | Preflight feature discovery. See [`11-capabilities-versioning.md`](./11-capabilities-versioning.md). |
|
||||
| `/workspace/preflight` | Route -> `DaemonStatusProvider` | Structured readiness cells: Node version, CLI entry, ripgrep, git, npm, plus ACP-level cells once a child is alive. |
|
||||
@@ -122,6 +122,22 @@ A **second** SIGTERM/SIGINT intentionally triggers `bridge.killAllSync()` + `pro
|
||||
|
||||
The status payload is daemon-only. `promptQueueWait` summarizes prompt FIFO queue wait samples observed in the daemon process. ACP child event loop lag is intentionally not aggregated into `/daemon/status`; it is visible through OTel gauge `qwen-code.acp.event_loop.lag` and through stderr stall lines forwarded into daemon logs.
|
||||
|
||||
### 10. Did file logging degrade or lose records?
|
||||
|
||||
Use full daemon status:
|
||||
|
||||
```bash
|
||||
curl -s 'http://127.0.0.1:4170/daemon/status?detail=full' | \
|
||||
jq '{status, issues, daemon: {runId: .daemon.runId, logMode: .daemon.logMode, logHealth: .daemon.logHealth, logPath: .daemon.logPath, logIssues: .daemon.logIssues, droppedRecords: .daemon.logDroppedRecords, droppedBytes: .daemon.logDroppedBytes}}'
|
||||
```
|
||||
|
||||
`stable` is the normal owner, `fallback` means another daemon owns the stable
|
||||
family, and `stderr-only` means file logging is disabled or unavailable.
|
||||
`fallback/ok` is expected under intentional concurrency. A
|
||||
`daemon_log_degraded` warning contains no path; request full detail for the
|
||||
actual path and logger issue codes. Use `runId` to separate restarts inside the
|
||||
stable file.
|
||||
|
||||
New OTel metric names:
|
||||
|
||||
- `qwen-code.daemon.event_loop.lag`, gauge in milliseconds with `stat=mean|p50|p99|max`.
|
||||
@@ -166,13 +182,16 @@ flowchart TD
|
||||
| ------------------------------- | -------------------------------------------------------------------------------------------- |
|
||||
| `QWEN_SERVE_DEBUG` | Enables verbose stderr logs. See [`17-configuration.md`](./17-configuration.md). |
|
||||
| `settings.json` `telemetry` | Controls OTel behavior: `enabled`, `otlpEndpoint`, `otlpProtocol`, and per-signal endpoints. |
|
||||
| `DaemonLogger` log path | Generated at boot and printed to stderr as `daemon log -> <path>`. |
|
||||
| `DaemonLogger` log path | Stable `debug/daemon/daemon.log`, or a run-specific fallback selected at boot. |
|
||||
| `PermissionAuditRing` size | Hard-coded to 512 today. |
|
||||
| `slow_client_warning` threshold | `0.75` / `0.375`, hard-coded in `eventBus.ts`. |
|
||||
|
||||
## Caveats and known limits
|
||||
|
||||
- **DaemonLogger file logs are structured** and can be filtered by `route`, `sessionId`, and `clientId`. `QWEN_SERVE_DEBUG` stderr logs remain unstructured text.
|
||||
- **DaemonLogger retention is size based, not age based.** The active file and four archives are bounded per family; live fallback owners are never deleted.
|
||||
- **Access summaries are intentional loss accounting.** A WARN `access logs suppressed` represents individual access records omitted from both stderr and file; it does not indicate dropped HTTP requests.
|
||||
- **External logrotate must not mutate the active family.** Use a shipper that reads/copies and reopens the stable pathname after replacement.
|
||||
- **OpenTelemetry spans include per-request correlation.** Classified daemon API requests that pass bearer authentication, rate limiting, and body parsing carry canonical route, sessionId, clientId, and (when uniquely resolved) `qwen-code.workspace.hash` attributes. Requests rejected by an earlier middleware gate do not have these request spans.
|
||||
- **HTTP metrics are daemon-global.** OpenTelemetry HTTP request metrics and the Web Shell status metrics ring do not include a workspace dimension. A successful session SSE connection has a request span but is excluded from ordinary request count/duration metrics because its lifetime is not request latency; failed SSE handshakes are counted normally.
|
||||
- **`runtime.perf` is daemon-only.** Child event loop lag is not reported there by design; use OTel or forwarded stderr stall warnings for ACP child stalls.
|
||||
|
||||
@@ -737,8 +737,12 @@ stdout/stderr are forwarded into the daemon log with bearer tokens, sensitive
|
||||
worker environment values, and proxy URL credentials redacted.
|
||||
|
||||
Security: the response never includes bearer tokens, client ids, full ACP
|
||||
connection ids, device-flow user codes, or verification URLs. `summary` omits
|
||||
the daemon log path; `full` may include it for authenticated operators.
|
||||
connection ids, device-flow user codes, or verification URLs. Both detail
|
||||
levels may include additive `daemon.runId`, `daemon.logMode`, and
|
||||
`daemon.logHealth`. `summary` omits the daemon log path and loss details;
|
||||
`full` may include `logPath`, `logIssues`, `logDroppedRecords`, and
|
||||
`logDroppedBytes` for authenticated operators. Degraded file logging adds the
|
||||
path-free `daemon_log_degraded` warning to the normal status rollup.
|
||||
|
||||
### `GET /capabilities`
|
||||
|
||||
|
||||
@@ -712,16 +712,48 @@ const result = await flow.awaitCompletion({ signal: abortCtrl.signal });
|
||||
|
||||
## Daemon log file
|
||||
|
||||
`qwen serve` writes a per-process diagnostic log to:
|
||||
`qwen serve` appends diagnostic records across normal restarts at the stable
|
||||
active path:
|
||||
|
||||
```
|
||||
${QWEN_RUNTIME_DIR or ~/.qwen}/debug/daemon/serve-<pid>-<workspaceHash>.log
|
||||
${QWEN_RUNTIME_DIR or ~/.qwen}/debug/daemon/daemon.log
|
||||
```
|
||||
|
||||
A `latest` symlink in the same directory always points at the current process's log, so `tail -f ~/.qwen/debug/daemon/latest` will follow whichever daemon is running.
|
||||
Every file record includes a random per-start `runId` and the daemon PID. A
|
||||
successful stable owner also updates `debug/daemon/latest` to `daemon.log` on
|
||||
platforms that support symlinks. On macOS/Linux, follow rotation with:
|
||||
|
||||
```bash
|
||||
tail -F ~/.qwen/debug/daemon/daemon.log
|
||||
```
|
||||
|
||||
On other platforms, configure the viewer to reopen the pathname after it is
|
||||
replaced. A viewer that keeps only the old file handle will remain on the
|
||||
archive after rotation.
|
||||
|
||||
The log captures lifecycle messages, route errors (with `route=` and `sessionId=` context), ACP child stderr, and — when `QWEN_SERVE_DEBUG=1` is set — extra bridge breadcrumbs. Lines that go to stderr today still go to stderr; the file log is **additive**, not a replacement.
|
||||
|
||||
The active file rotates before it would exceed 10 MiB. Each family retains
|
||||
four archives under `archive/`, and each file record is capped at 256 KiB. The
|
||||
in-memory queue accepts at most 4 MiB of unsettled file payload. Queue pressure,
|
||||
rotation failures, or filesystem failures can therefore drop file copies;
|
||||
`GET /daemon/status?detail=full` exposes logger health, issues, and dropped
|
||||
record/byte counters.
|
||||
|
||||
Only one daemon may own the stable family in a log namespace. A concurrent
|
||||
daemon writes to `debug/daemon/runs/run-<runId>/daemon.log`; the startup banner
|
||||
and full status contain the authoritative path. `runs/recent-fallback` is a
|
||||
best-effort locator for a recent fallback family and may point to one that is
|
||||
still live. A healthy namespace converges to roughly 100 MiB: about 50 MiB for
|
||||
stable plus one inactive fallback family. Live or not-yet-stale fallback
|
||||
families are retained, so concurrent daemons or crash/restart storms can
|
||||
temporarily use more.
|
||||
|
||||
One runtime directory is one ownership and retention namespace. Use distinct
|
||||
`QWEN_RUNTIME_DIR` values when daemons need independent history. New daemon log
|
||||
directories are private to the user (`0700`) and new files use `0600` on POSIX.
|
||||
There is no age-based expiry.
|
||||
|
||||
### Disabling
|
||||
|
||||
Set `QWEN_DAEMON_LOG_FILE=0` (or `false`/`off`/`no`) to skip file logging entirely. Stderr output is unaffected.
|
||||
@@ -730,9 +762,14 @@ Set `QWEN_DAEMON_LOG_FILE=0` (or `false`/`off`/`no`) to skip file logging entire
|
||||
|
||||
Session-scoped debug logs (`~/.qwen/debug/<sessionId>.txt` and the `~/.qwen/debug/latest` symlink) are independent. The daemon log lives in a sibling `daemon/` subdirectory; per-session debug semantics are unchanged by this feature.
|
||||
|
||||
### No rotation
|
||||
### External rotation
|
||||
|
||||
The daemon log appends indefinitely. Rotate manually if it grows large. A future enhancement may add automatic rotation; track via [#4548](https://github.com/QwenLM/qwen-code/issues/4548) follow-ups.
|
||||
Do not point an external logrotate rule at the active `daemon.log`. The daemon
|
||||
is the sole supported writer and rotator; external rename, deletion, or
|
||||
truncation invalidates its size model. Copying or shipping records without
|
||||
mutating the family is safe. Older `serve-<pid>.log` and
|
||||
`serve-<pid>-<workspaceHash>.log` files are left untouched and are not counted
|
||||
by the new retention policy.
|
||||
|
||||
## Runtime MCP server management (issue [#4514](https://github.com/QwenLM/qwen-code/issues/4514))
|
||||
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
File diff suppressed because it is too large.
Load diff
@@ -24,6 +24,7 @@ import {
|
||||
import type { ChannelWorkerSnapshot } from './channel-worker-supervisor.js';
|
||||
import type { RateLimiterInstance, RateLimitTier } from './rate-limit.js';
|
||||
import type { DaemonWorkspaceService } from './workspace-service/index.js';
|
||||
import type { DaemonLogger } from './daemon-logger.js';
|
||||
|
||||
const BASE_WORKSPACE = '/work/status';
|
||||
|
||||
@@ -48,6 +49,76 @@ afterEach(() => {
|
||||
});
|
||||
|
||||
describe('buildDaemonStatusResponse', () => {
|
||||
it('uses one logger snapshot and exposes summary/full log diagnostics', async () => {
|
||||
const getStatus = vi.fn(() => ({
|
||||
runId: '0123456789abcdef0123456789abcdef',
|
||||
mode: 'stable' as const,
|
||||
health: 'ok' as const,
|
||||
issues: [] as const,
|
||||
droppedRecords: 2,
|
||||
droppedBytes: 42,
|
||||
}));
|
||||
const daemonLog = {
|
||||
getStatus,
|
||||
getDaemonId: () => 'daemon:123',
|
||||
getLogPath: () => '/runtime/debug/daemon/daemon.log',
|
||||
} as unknown as DaemonLogger;
|
||||
|
||||
const summary = await buildDaemonStatusResponse(
|
||||
'summary',
|
||||
makeOptions({ daemonLog }),
|
||||
);
|
||||
expect(getStatus).toHaveBeenCalledOnce();
|
||||
expect(summary.daemon).toMatchObject({
|
||||
runId: '0123456789abcdef0123456789abcdef',
|
||||
logMode: 'stable',
|
||||
logHealth: 'ok',
|
||||
});
|
||||
expect(summary.daemon).not.toHaveProperty('logPath');
|
||||
expect(summary.daemon).not.toHaveProperty('logIssues');
|
||||
|
||||
getStatus.mockClear();
|
||||
const full = await buildDaemonStatusResponse(
|
||||
'full',
|
||||
makeOptions({ daemonLog }),
|
||||
);
|
||||
expect(getStatus).toHaveBeenCalledOnce();
|
||||
expect(full.daemon).toMatchObject({
|
||||
logPath: '/runtime/debug/daemon/daemon.log',
|
||||
logIssues: [],
|
||||
logDroppedRecords: 2,
|
||||
logDroppedBytes: 42,
|
||||
});
|
||||
});
|
||||
|
||||
it('rolls degraded logger health into a path-free warning', async () => {
|
||||
const daemonLog = {
|
||||
getStatus: () => ({
|
||||
runId: '0123456789abcdef0123456789abcdef',
|
||||
mode: 'stderr-only' as const,
|
||||
health: 'degraded' as const,
|
||||
issues: ['init_failed'] as const,
|
||||
droppedRecords: 0,
|
||||
droppedBytes: 0,
|
||||
}),
|
||||
getDaemonId: () => 'daemon:123',
|
||||
getLogPath: () => '/secret/path',
|
||||
} as unknown as DaemonLogger;
|
||||
const response = await buildDaemonStatusResponse(
|
||||
'summary',
|
||||
makeOptions({ daemonLog }),
|
||||
);
|
||||
|
||||
expect(response.status).toBe('warning');
|
||||
expect(response.issues).toContainEqual({
|
||||
code: 'daemon_log_degraded',
|
||||
severity: 'warning',
|
||||
message:
|
||||
'Daemon file logging is degraded; inspect full status for details.',
|
||||
});
|
||||
expect(JSON.stringify(response.issues)).not.toContain('/secret/path');
|
||||
});
|
||||
|
||||
it('includes maxTotalSessions in daemon status limits', async () => {
|
||||
const options = makeOptions();
|
||||
options.opts.maxTotalSessions = 50;
|
||||
@@ -1029,6 +1100,7 @@ interface MakeOptionsInput {
|
||||
lastActivityAt?: number | null;
|
||||
totalAdmissionLiveCount?: number;
|
||||
totalAdmissionInFlight?: number;
|
||||
daemonLog?: DaemonLogger;
|
||||
}
|
||||
|
||||
function makeOptions(input: MakeOptionsInput = {}): BuildDaemonStatusOptions {
|
||||
@@ -1070,6 +1142,7 @@ function makeOptions(input: MakeOptionsInput = {}): BuildDaemonStatusOptions {
|
||||
bridge,
|
||||
workspace,
|
||||
qwenCodeVersion: 'test',
|
||||
daemonLog: input.daemonLog,
|
||||
...(input.acpSnapshot
|
||||
? {
|
||||
acpHandle: {
|
||||
|
||||
@@ -7,7 +7,12 @@
|
||||
import type { ServeProtocolVersions } from './capabilities.js';
|
||||
import type { AcpHttpHandle, AcpHttpSnapshot } from './acp-http/index.js';
|
||||
import type { DeviceFlowRegistry } from './auth/device-flow.js';
|
||||
import type { DaemonLogger } from './daemon-logger.js';
|
||||
import type {
|
||||
DaemonLogger,
|
||||
DaemonLogHealth,
|
||||
DaemonLogIssue,
|
||||
DaemonLogMode,
|
||||
} from './daemon-logger.js';
|
||||
import type {
|
||||
AcpSessionBridge,
|
||||
BridgeDaemonStatusSnapshot,
|
||||
@@ -76,7 +81,8 @@ export interface DaemonStatusIssue {
|
||||
| 'channel_worker_exited'
|
||||
| 'channel_worker_partial_connect'
|
||||
| 'daemon_runtime_starting'
|
||||
| 'daemon_runtime_failed';
|
||||
| 'daemon_runtime_failed'
|
||||
| 'daemon_log_degraded';
|
||||
severity: IssueSeverity;
|
||||
message: string;
|
||||
section?: string;
|
||||
@@ -249,6 +255,12 @@ export interface DaemonStatusResponse {
|
||||
uptimeMs: number;
|
||||
mode: ServeOptions['mode'];
|
||||
workspaceCwd: string;
|
||||
runId?: string;
|
||||
logMode?: DaemonLogMode;
|
||||
logHealth?: DaemonLogHealth;
|
||||
logIssues?: readonly DaemonLogIssue[];
|
||||
logDroppedRecords?: number;
|
||||
logDroppedBytes?: number;
|
||||
};
|
||||
security: DaemonStatusSecurity;
|
||||
limits: DaemonStatusLimits;
|
||||
@@ -290,6 +302,7 @@ export async function buildDaemonStatusResponse(
|
||||
detail: DaemonStatusDetail,
|
||||
input: BuildDaemonStatusOptions,
|
||||
): Promise<DaemonStatusResponse> {
|
||||
const daemonLogStatus = input.daemonLog?.getStatus();
|
||||
const bridgeSnapshot = input.bridge.getDaemonStatusSnapshot();
|
||||
const lastActivity = input.bridge.lastActivityAt ?? null;
|
||||
const workspaceRuntimes = input.workspaceRegistry?.list();
|
||||
@@ -389,6 +402,14 @@ export async function buildDaemonStatusResponse(
|
||||
totalAdmissionSnapshot,
|
||||
workspaceSnapshots,
|
||||
);
|
||||
if (daemonLogStatus?.health === 'degraded') {
|
||||
issues.push({
|
||||
code: 'daemon_log_degraded',
|
||||
severity: 'warning',
|
||||
message:
|
||||
'Daemon file logging is degraded; inspect full status for details.',
|
||||
});
|
||||
}
|
||||
|
||||
if (detail === 'full') {
|
||||
full = await buildFullStatus(
|
||||
@@ -417,9 +438,23 @@ export async function buildDaemonStatusResponse(
|
||||
...(input.daemonLog?.getDaemonId()
|
||||
? { daemonId: input.daemonLog.getDaemonId() }
|
||||
: {}),
|
||||
...(daemonLogStatus
|
||||
? {
|
||||
runId: daemonLogStatus.runId,
|
||||
logMode: daemonLogStatus.mode,
|
||||
logHealth: daemonLogStatus.health,
|
||||
}
|
||||
: {}),
|
||||
...(detail === 'full' && input.daemonLog?.getLogPath()
|
||||
? { logPath: input.daemonLog.getLogPath() }
|
||||
: {}),
|
||||
...(detail === 'full' && daemonLogStatus
|
||||
? {
|
||||
logIssues: daemonLogStatus.issues,
|
||||
logDroppedRecords: daemonLogStatus.droppedRecords,
|
||||
logDroppedBytes: daemonLogStatus.droppedBytes,
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
security: {
|
||||
tokenConfigured: Boolean(input.opts.token),
|
||||
|
||||
@@ -713,7 +713,16 @@ function makeDaemonLog(): DaemonLogger {
|
||||
raw: vi.fn(),
|
||||
getLogPath: () => '',
|
||||
getDaemonId: () => 'test-daemon',
|
||||
getStatus: () => ({
|
||||
runId: '00000000000000000000000000000000',
|
||||
mode: 'stderr-only',
|
||||
health: 'ok',
|
||||
issues: [],
|
||||
droppedRecords: 0,
|
||||
droppedBytes: 0,
|
||||
}),
|
||||
flush: vi.fn(async () => {}),
|
||||
close: vi.fn(async () => {}),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ import * as fs from 'node:fs';
|
||||
import { createServer } from 'node:http';
|
||||
import * as https from 'node:https';
|
||||
import type { AddressInfo } from 'node:net';
|
||||
import { describe, it, expect, vi, afterEach } from 'vitest';
|
||||
import { describe, it, expect, vi, afterEach, afterAll } from 'vitest';
|
||||
import express from 'express';
|
||||
import {
|
||||
createLazyBridgeProxy,
|
||||
@@ -55,6 +55,25 @@ import {
|
||||
} from './workspace-registration-store.js';
|
||||
import { getDeferredRuntimeRequestTiming } from './server/request-helpers.js';
|
||||
|
||||
const originalTestRuntimeDir = process.env['QWEN_RUNTIME_DIR'];
|
||||
const isolatedTestRuntimeDir = fs.realpathSync(
|
||||
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-run-serve-tests-')),
|
||||
);
|
||||
process.env['QWEN_RUNTIME_DIR'] = isolatedTestRuntimeDir;
|
||||
|
||||
afterEach(() => {
|
||||
process.env['QWEN_RUNTIME_DIR'] = isolatedTestRuntimeDir;
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
if (originalTestRuntimeDir === undefined) {
|
||||
delete process.env['QWEN_RUNTIME_DIR'];
|
||||
} else {
|
||||
process.env['QWEN_RUNTIME_DIR'] = originalTestRuntimeDir;
|
||||
}
|
||||
fs.rmSync(isolatedTestRuntimeDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
const BASE_BRIDGE_SNAPSHOT: BridgeDaemonStatusSnapshot = {
|
||||
limits: {
|
||||
maxSessions: 20,
|
||||
@@ -451,14 +470,14 @@ describe('runQwenServe daemon logger wiring', () => {
|
||||
const daemonDir = path.join(debugDir, 'daemon');
|
||||
expect(fs.existsSync(daemonDir)).toBe(true);
|
||||
|
||||
// Find the daemon-scoped log file.
|
||||
// Find the stable daemon log file.
|
||||
const logFiles = fs
|
||||
.readdirSync(daemonDir)
|
||||
.filter((f) => f.endsWith('.log'));
|
||||
expect(logFiles).toContain(`serve-${process.pid}.log`);
|
||||
expect(logFiles).toContain('daemon.log');
|
||||
|
||||
const logContent = fs.readFileSync(
|
||||
path.join(daemonDir, `serve-${process.pid}.log`),
|
||||
path.join(daemonDir, 'daemon.log'),
|
||||
'utf8',
|
||||
);
|
||||
// Should contain the "daemon started" boot line
|
||||
@@ -468,22 +487,39 @@ describe('runQwenServe daemon logger wiring', () => {
|
||||
`workspace=${fs.realpathSync.native(workspace)}`,
|
||||
);
|
||||
|
||||
await Promise.all(
|
||||
Array.from({ length: 70 }, (_, index) =>
|
||||
fetch(`${handle.url}/missing-${index}`),
|
||||
),
|
||||
);
|
||||
|
||||
// Close the handle (graceful shutdown)
|
||||
await handle.close();
|
||||
|
||||
// The log should still be readable after shutdown
|
||||
const finalContent = fs.readFileSync(
|
||||
path.join(daemonDir, `serve-${process.pid}.log`),
|
||||
'utf8',
|
||||
// close() is intentionally bounded, so the file finalizer may still be
|
||||
// draining when it returns under a slow filesystem.
|
||||
const logPath = path.join(daemonDir, 'daemon.log');
|
||||
let finalContent = '';
|
||||
await vi.waitFor(
|
||||
() => {
|
||||
finalContent = fs.readFileSync(logPath, 'utf8');
|
||||
expect(finalContent).toContain('access logs suppressed');
|
||||
expect(finalContent).toContain('daemon stopped');
|
||||
},
|
||||
{ timeout: 7_000, interval: 50 },
|
||||
);
|
||||
expect(finalContent).toContain('daemon started');
|
||||
const suppressedIndex = finalContent.indexOf('access logs suppressed');
|
||||
const stoppedIndex = finalContent.indexOf('daemon stopped');
|
||||
expect(suppressedIndex).toBeGreaterThanOrEqual(0);
|
||||
expect(stoppedIndex).toBeGreaterThan(suppressedIndex);
|
||||
} finally {
|
||||
delete process.env['QWEN_RUNTIME_DIR'];
|
||||
if (origEnv !== undefined) {
|
||||
process.env['QWEN_RUNTIME_DIR'] = origEnv;
|
||||
}
|
||||
}
|
||||
});
|
||||
}, 10_000);
|
||||
});
|
||||
|
||||
describe('runQwenServe telemetry validation', () => {
|
||||
@@ -1181,6 +1217,12 @@ describe('runQwenServe permissionResponseTimeoutMs validation', () => {
|
||||
{ bridge: fakeBridge },
|
||||
),
|
||||
).rejects.toThrow(/permissionResponseTimeoutMs/);
|
||||
const log = fs.readFileSync(
|
||||
path.join(tmpDir, 'debug', 'daemon', 'daemon.log'),
|
||||
'utf8',
|
||||
);
|
||||
expect(log).toContain('daemon startup failed');
|
||||
expect(log).not.toContain('daemon stopped');
|
||||
} finally {
|
||||
delete process.env['QWEN_RUNTIME_DIR'];
|
||||
if (origEnv !== undefined) {
|
||||
@@ -1188,6 +1230,53 @@ describe('runQwenServe permissionResponseTimeoutMs validation', () => {
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it('preserves the startup error and releases the log lease when stderr fails', async () => {
|
||||
tmpDir = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), 'qws-pt-')));
|
||||
const fakeBridge = {
|
||||
spawnOrAttach: vi.fn(),
|
||||
shutdown: vi.fn().mockResolvedValue(undefined),
|
||||
killAllSync: vi.fn(),
|
||||
} as unknown as HttpAcpBridge;
|
||||
const origEnv = process.env['QWEN_RUNTIME_DIR'];
|
||||
process.env['QWEN_RUNTIME_DIR'] = tmpDir;
|
||||
const stderr = vi
|
||||
.spyOn(process.stderr, 'write')
|
||||
.mockImplementation((chunk) => {
|
||||
if (String(chunk).includes('daemon startup failed')) {
|
||||
throw new Error('stderr unavailable');
|
||||
}
|
||||
return true;
|
||||
});
|
||||
|
||||
try {
|
||||
await expect(
|
||||
runQwenServe(
|
||||
{
|
||||
port: 0,
|
||||
hostname: '127.0.0.1',
|
||||
mode: 'http-bridge',
|
||||
workspace: tmpDir,
|
||||
maxSessions: 1,
|
||||
permissionResponseTimeoutMs: Number.NaN,
|
||||
},
|
||||
{ bridge: fakeBridge },
|
||||
),
|
||||
).rejects.toThrow(/permissionResponseTimeoutMs/);
|
||||
} finally {
|
||||
stderr.mockRestore();
|
||||
delete process.env['QWEN_RUNTIME_DIR'];
|
||||
if (origEnv !== undefined) {
|
||||
process.env['QWEN_RUNTIME_DIR'] = origEnv;
|
||||
}
|
||||
}
|
||||
|
||||
expect(
|
||||
fs.existsSync(
|
||||
path.join(tmpDir, 'debug', 'daemon', '.stable-writer.lock'),
|
||||
),
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
// Long-lived self-signed cert (CN=localhost, SAN IP:127.0.0.1) used only
|
||||
@@ -2261,11 +2350,7 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
|
||||
await handle.close();
|
||||
closed = true;
|
||||
const logPath = path.join(
|
||||
logBaseDir,
|
||||
'daemon',
|
||||
`serve-${process.pid}.log`,
|
||||
);
|
||||
const logPath = path.join(logBaseDir, 'daemon', 'daemon.log');
|
||||
const log = fs.readFileSync(logPath, 'utf8');
|
||||
expect(log).toContain(
|
||||
'failed to rebuild runtime env snapshot after daemon env reload; preserving previous runtime env',
|
||||
@@ -3710,7 +3795,7 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
closed = true;
|
||||
|
||||
const log = fs.readFileSync(
|
||||
path.join(logBaseDir, 'daemon', `serve-${process.pid}.log`),
|
||||
path.join(logBaseDir, 'daemon', 'daemon.log'),
|
||||
'utf8',
|
||||
);
|
||||
expect(log).toContain('deferred webhook auth failed');
|
||||
@@ -4701,6 +4786,8 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
delete process.env['QWEN_SERVE_CLIENT_MCP_OVER_WS'];
|
||||
delete process.env['QWEN_SERVE_CDP_TUNNEL_OVER_WS'];
|
||||
const boundWorkspace = canonicalizeWorkspace(tmpDir);
|
||||
const blockedLogBaseDir = path.join(tmpDir, 'blocked-log-base');
|
||||
fs.writeFileSync(blockedLogBaseDir, 'not a directory');
|
||||
vi.spyOn(acpBridge, 'createAcpSessionBridge').mockImplementation(() => {
|
||||
throw new Error('runtime boom');
|
||||
});
|
||||
@@ -4714,7 +4801,7 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
maxSessions: 1,
|
||||
serveWebShell: false,
|
||||
},
|
||||
{ resolveOnListen: true },
|
||||
{ resolveOnListen: true, daemonLogBaseDir: blockedLogBaseDir },
|
||||
);
|
||||
|
||||
try {
|
||||
@@ -4780,6 +4867,11 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
const body = (await res.json()) as {
|
||||
status?: string;
|
||||
issues?: Array<{ code?: string; severity?: string }>;
|
||||
daemon?: {
|
||||
runId?: string;
|
||||
logMode?: string;
|
||||
logHealth?: string;
|
||||
};
|
||||
runtime?: { loading?: boolean; error?: string };
|
||||
};
|
||||
expect(body).toMatchObject({
|
||||
@@ -4789,7 +4881,16 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
code: 'daemon_runtime_failed',
|
||||
severity: 'error',
|
||||
}),
|
||||
expect.objectContaining({
|
||||
code: 'daemon_log_degraded',
|
||||
severity: 'warning',
|
||||
}),
|
||||
]),
|
||||
daemon: {
|
||||
runId: expect.stringMatching(/^[0-9a-f]{32}$/),
|
||||
logMode: 'stderr-only',
|
||||
logHealth: 'degraded',
|
||||
},
|
||||
runtime: { loading: false, error: 'runtime boom' },
|
||||
});
|
||||
|
||||
@@ -4804,6 +4905,20 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
expect(sameOriginBody).toMatchObject({
|
||||
v: 1,
|
||||
detail: 'full',
|
||||
issues: expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
code: 'daemon_log_degraded',
|
||||
severity: 'warning',
|
||||
}),
|
||||
]),
|
||||
daemon: {
|
||||
runId: body.daemon?.runId,
|
||||
logMode: 'stderr-only',
|
||||
logHealth: 'degraded',
|
||||
logIssues: ['init_failed'],
|
||||
logDroppedRecords: 0,
|
||||
logDroppedBytes: 0,
|
||||
},
|
||||
security: { allowOriginMode: 'none' },
|
||||
limits: {
|
||||
maxSessions: 1,
|
||||
@@ -4846,6 +4961,7 @@ describe('runQwenServe runtime startup failures', () => {
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(sameOriginBody.daemon).not.toHaveProperty('logPath');
|
||||
} finally {
|
||||
if (originalClientMcpOverWs === undefined) {
|
||||
delete process.env['QWEN_SERVE_CLIENT_MCP_OVER_WS'];
|
||||
@@ -6697,6 +6813,7 @@ describe('runQwenServe channel worker supervisor', () => {
|
||||
bridge,
|
||||
channelWorkerSupervisorFactory: vi.fn(() => worker),
|
||||
channelServicePidfile: pidfile,
|
||||
daemonLogBaseDir: path.join(tmpDir, 'debug'),
|
||||
},
|
||||
);
|
||||
|
||||
@@ -6714,6 +6831,8 @@ describe('runQwenServe channel worker supervisor', () => {
|
||||
expect(exitSpy).not.toHaveBeenCalled();
|
||||
expect(worker.killAllSync).not.toHaveBeenCalled();
|
||||
expect(pidfile.removeServeServiceInfo).not.toHaveBeenCalled();
|
||||
const logPath = path.join(tmpDir, 'debug', 'daemon', 'daemon.log');
|
||||
expect(fs.readFileSync(logPath, 'utf8')).not.toContain('daemon stopped');
|
||||
|
||||
await signalListener!('SIGTERM');
|
||||
expect(worker.stop).toHaveBeenCalledTimes(2);
|
||||
@@ -6721,6 +6840,7 @@ describe('runQwenServe channel worker supervisor', () => {
|
||||
expect(bridge.killAllSync).not.toHaveBeenCalled();
|
||||
expect(pidfile.removeServeServiceInfo).toHaveBeenCalledWith(process.pid);
|
||||
expect(exitSpy).toHaveBeenCalledWith(0);
|
||||
expect(fs.readFileSync(logPath, 'utf8')).toContain('daemon stopped');
|
||||
} finally {
|
||||
for (const listener of process.rawListeners('SIGINT')) {
|
||||
if (!existingSigintListeners.has(listener)) {
|
||||
@@ -6736,6 +6856,92 @@ describe('runQwenServe channel worker supervisor', () => {
|
||||
}
|
||||
});
|
||||
|
||||
it('bounds the logger flush before allowing a retryable close to reject', async () => {
|
||||
tmpDir = fs.realpathSync(
|
||||
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-channel-log-stuck-')),
|
||||
);
|
||||
const bridge = makeFakeBridge();
|
||||
const worker = makeWorker({
|
||||
enabled: true,
|
||||
state: 'failed',
|
||||
pid: 1234,
|
||||
channels: ['telegram'],
|
||||
error: 'Channel worker did not exit after SIGKILL.',
|
||||
});
|
||||
worker.stop
|
||||
.mockRejectedValueOnce(
|
||||
new Error('Channel worker did not exit after SIGKILL.'),
|
||||
)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const pidfile = makePidfileDeps();
|
||||
const logPath = path.join(tmpDir, 'debug', 'daemon', 'daemon.log');
|
||||
const handle = await runQwenServe(
|
||||
{
|
||||
port: 0,
|
||||
hostname: '127.0.0.1',
|
||||
mode: 'http-bridge',
|
||||
workspace: tmpDir,
|
||||
serveWebShell: false,
|
||||
channelSelection: { mode: 'names', names: ['telegram'] },
|
||||
},
|
||||
{
|
||||
bridge,
|
||||
channelWorkerSupervisorFactory: vi.fn(() => worker),
|
||||
channelServicePidfile: pidfile,
|
||||
daemonLogBaseDir: path.join(tmpDir, 'debug'),
|
||||
},
|
||||
);
|
||||
|
||||
let releaseAppend!: () => void;
|
||||
const appendGate = new Promise<void>((resolve) => {
|
||||
releaseAppend = resolve;
|
||||
});
|
||||
const originalAppendFile = fs.promises.appendFile.bind(fs.promises);
|
||||
const appendSpy = vi
|
||||
.spyOn(fs.promises, 'appendFile')
|
||||
.mockImplementation(async (...args) => {
|
||||
if (String(args[0]) === logPath) await appendGate;
|
||||
return originalAppendFile(...args);
|
||||
});
|
||||
let closeOutcome:
|
||||
| Promise<{ kind: 'resolved' } | { kind: 'rejected'; error: unknown }>
|
||||
| undefined;
|
||||
|
||||
try {
|
||||
const response = await fetch(`${handle.url}/blocked-access-log`);
|
||||
await response.text();
|
||||
closeOutcome = handle.close().then(
|
||||
() => ({ kind: 'resolved' as const }),
|
||||
(error: unknown) => ({ kind: 'rejected' as const, error }),
|
||||
);
|
||||
let timeout: NodeJS.Timeout | undefined;
|
||||
const firstOutcome = await Promise.race([
|
||||
closeOutcome,
|
||||
new Promise<{ kind: 'timeout' }>((resolve) => {
|
||||
timeout = setTimeout(() => resolve({ kind: 'timeout' }), 1_500);
|
||||
}),
|
||||
]);
|
||||
if (timeout) clearTimeout(timeout);
|
||||
|
||||
expect(firstOutcome.kind).toBe('rejected');
|
||||
if (firstOutcome.kind === 'rejected') {
|
||||
expect(firstOutcome.error).toEqual(
|
||||
expect.objectContaining({
|
||||
message: 'Channel worker did not exit after SIGKILL.',
|
||||
}),
|
||||
);
|
||||
}
|
||||
} finally {
|
||||
releaseAppend();
|
||||
appendSpy.mockRestore();
|
||||
await closeOutcome;
|
||||
await handle.close();
|
||||
}
|
||||
|
||||
expect(worker.stop).toHaveBeenCalledTimes(2);
|
||||
expect(fs.readFileSync(logPath, 'utf8')).toContain('daemon stopped');
|
||||
});
|
||||
|
||||
it('retries bridge shutdown when channel and bridge teardown fail together', async () => {
|
||||
tmpDir = fs.realpathSync(
|
||||
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-channel-bridge-stuck-')),
|
||||
|
||||
@@ -141,6 +141,10 @@ import type {
|
||||
import { sanitizeLogText } from '@qwen-code/channel-base';
|
||||
import { isBrowserAutomationMcpAvailable } from './cdp-mcp-command.js';
|
||||
import { WorkspaceVoiceCoordinator } from './voice/workspace-voice-coordinator.js';
|
||||
import {
|
||||
ACCESS_LOG_CONTROLLER_LOCAL,
|
||||
type AccessLogAppLocals,
|
||||
} from './server/access-log.js';
|
||||
import {
|
||||
setDeferredRuntimeRequestTiming,
|
||||
type DeferredRuntimeRequestTiming,
|
||||
@@ -154,6 +158,33 @@ const QWEN_SERVE_PROMPT_DEADLINE_MS_ENV = 'QWEN_SERVE_PROMPT_DEADLINE_MS';
|
||||
const QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS_ENV =
|
||||
'QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS';
|
||||
const SHUTDOWN_FORCE_CLOSE_MS = 5_000;
|
||||
const DAEMON_LOG_FORCED_FLUSH_BUDGET_MS = 250;
|
||||
|
||||
async function flushDaemonLogBounded(
|
||||
daemonLog: DaemonLogger,
|
||||
budgetMs: number,
|
||||
): Promise<void> {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
try {
|
||||
await Promise.race([
|
||||
daemonLog.flush().catch(() => {}),
|
||||
new Promise<void>((resolve) => {
|
||||
timer = setTimeout(resolve, budgetMs);
|
||||
timer.unref();
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
function writeDaemonLifecycleBestEffort(write: () => void): void {
|
||||
try {
|
||||
write();
|
||||
} catch {
|
||||
// Best-effort lifecycle diagnostics must not make shutdown throw.
|
||||
}
|
||||
}
|
||||
|
||||
function daemonPipeDirection(
|
||||
direction: NdJsonMessageObservation['direction'],
|
||||
@@ -1297,12 +1328,22 @@ function createBootstrapServeApp(input: {
|
||||
severity: 'warning',
|
||||
message: 'Daemon runtime is still starting.',
|
||||
};
|
||||
const daemonLogStatus = daemonLog.getStatus();
|
||||
const issues: DaemonStatusIssue[] = [issue];
|
||||
if (daemonLogStatus.health === 'degraded') {
|
||||
issues.push({
|
||||
code: 'daemon_log_degraded',
|
||||
severity: 'warning',
|
||||
message:
|
||||
'Daemon file logging is degraded; inspect full status for details.',
|
||||
});
|
||||
}
|
||||
const response: DaemonStatusResponse = {
|
||||
v: 1,
|
||||
detail: detail.detail,
|
||||
generatedAt: new Date().toISOString(),
|
||||
status: runtimeFailed ? 'error' : 'warning',
|
||||
issues: [issue],
|
||||
issues,
|
||||
daemon: {
|
||||
pid: process.pid,
|
||||
uptimeMs: Math.round(process.uptime() * 1000),
|
||||
@@ -1316,9 +1357,19 @@ function createBootstrapServeApp(input: {
|
||||
...(daemonLog.getDaemonId()
|
||||
? { daemonId: daemonLog.getDaemonId() }
|
||||
: {}),
|
||||
runId: daemonLogStatus.runId,
|
||||
logMode: daemonLogStatus.mode,
|
||||
logHealth: daemonLogStatus.health,
|
||||
...(detail.detail === 'full' && daemonLog.getLogPath()
|
||||
? { logPath: daemonLog.getLogPath() }
|
||||
: {}),
|
||||
...(detail.detail === 'full'
|
||||
? {
|
||||
logIssues: daemonLogStatus.issues,
|
||||
logDroppedRecords: daemonLogStatus.droppedRecords,
|
||||
logDroppedBytes: daemonLogStatus.droppedBytes,
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
security: {
|
||||
tokenConfigured: Boolean(opts.token),
|
||||
@@ -1638,9 +1689,49 @@ function runSynchronousRequestGate(
|
||||
* Boot refuses to start when bound beyond loopback without a token; this is a
|
||||
* hard rule, not a warning, per the threat model in the design issue.
|
||||
*/
|
||||
interface DaemonLoggerLifecycleCallbacks {
|
||||
initialized(logger: DaemonLogger): void;
|
||||
published(): void;
|
||||
signalOwned(): void;
|
||||
}
|
||||
|
||||
export async function runQwenServe(
|
||||
optsIn: RunQwenServeOptions,
|
||||
deps: RunQwenServeDeps = {},
|
||||
): Promise<RunHandle> {
|
||||
let daemonLog: DaemonLogger | undefined;
|
||||
let owner: 'startup' | 'handle' | 'signal' = 'startup';
|
||||
try {
|
||||
return await runQwenServeImpl(optsIn, deps, {
|
||||
initialized: (logger) => {
|
||||
daemonLog = logger;
|
||||
},
|
||||
published: () => {
|
||||
if (owner === 'startup') owner = 'handle';
|
||||
},
|
||||
signalOwned: () => {
|
||||
if (owner === 'startup') owner = 'signal';
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
if (daemonLog && owner === 'startup') {
|
||||
const startupLog = daemonLog;
|
||||
writeDaemonLifecycleBestEffort(() =>
|
||||
startupLog.error(
|
||||
'daemon startup failed',
|
||||
error instanceof Error ? error : new Error(String(error)),
|
||||
),
|
||||
);
|
||||
await startupLog.close();
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function runQwenServeImpl(
|
||||
optsIn: RunQwenServeOptions,
|
||||
deps: RunQwenServeDeps,
|
||||
loggerLifecycle: DaemonLoggerLifecycleCallbacks,
|
||||
): Promise<RunHandle> {
|
||||
const runStartedAt = performance.now();
|
||||
const shouldPreheat = !deps.bridge && shouldPreheatBridge(deps);
|
||||
@@ -2185,10 +2276,13 @@ export async function runQwenServe(
|
||||
bootSettings,
|
||||
boundWorkspace,
|
||||
});
|
||||
const daemonLog: DaemonLogger = initDaemonLogger({
|
||||
const daemonLog: DaemonLogger = await initDaemonLogger({
|
||||
boundWorkspace,
|
||||
baseDir: daemonLogBaseDir,
|
||||
});
|
||||
loggerLifecycle.initialized(daemonLog);
|
||||
let loggerPublished = false;
|
||||
let loggerSignalOwned = false;
|
||||
writeStderrLine(
|
||||
`qwen serve: daemon log → ${daemonLog.getLogPath() || '(disabled)'}`,
|
||||
);
|
||||
@@ -5081,25 +5175,33 @@ export async function runQwenServe(
|
||||
err instanceof Error ? err : null,
|
||||
);
|
||||
}
|
||||
await daemonLog.flush().catch(() => {});
|
||||
await flushDaemonLogBounded(
|
||||
daemonLog,
|
||||
DAEMON_LOG_FORCED_FLUSH_BUDGET_MS,
|
||||
);
|
||||
process.exit(1);
|
||||
return;
|
||||
}
|
||||
if (!loggerPublished) {
|
||||
loggerSignalOwned = true;
|
||||
loggerLifecycle.signalOwned();
|
||||
}
|
||||
daemonLog.warn(`received ${signal}, draining`);
|
||||
try {
|
||||
await handle.close();
|
||||
await daemonLog.flush();
|
||||
process.exit(runtimeStartupError === undefined ? 0 : 1);
|
||||
} catch (err) {
|
||||
daemonLog.error('shutdown error', err instanceof Error ? err : null);
|
||||
await daemonLog.flush().catch(() => {});
|
||||
if (channelWorkerManager?.state().enabled) {
|
||||
daemonLog.error(
|
||||
'refusing to exit while a channel worker or service lease remains; signal again to retry after the child exits (another signal during that retry forces exit)',
|
||||
);
|
||||
await daemonLog.flush().catch(() => {});
|
||||
return;
|
||||
}
|
||||
await flushDaemonLogBounded(
|
||||
daemonLog,
|
||||
DAEMON_LOG_FORCED_FLUSH_BUDGET_MS,
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
};
|
||||
@@ -5178,6 +5280,12 @@ export async function runQwenServe(
|
||||
const finish = (err?: Error | null) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
const accessLogController = (
|
||||
(runtimeApp ?? runtimeAppForCleanup)?.locals as
|
||||
| AccessLogAppLocals
|
||||
| undefined
|
||||
)?.[ACCESS_LOG_CONTROLLER_LOCAL];
|
||||
accessLogController?.sealAndFlushSuppressed();
|
||||
const preserveSignalHandlers =
|
||||
channelWorkerShutdownError !== undefined &&
|
||||
channelWorkerManager?.state().enabled === true;
|
||||
@@ -5203,26 +5311,34 @@ export async function runQwenServe(
|
||||
}`,
|
||||
);
|
||||
})
|
||||
.finally(() => daemonLog.flush().catch(() => {}))
|
||||
.finally(() => {
|
||||
.finally(async () => {
|
||||
// Server.close error takes precedence (operator-visible
|
||||
// listener problem); fall back to the bridge error
|
||||
// captured during shutdown if any.
|
||||
const finalErr =
|
||||
err ?? bridgeShutdownError ?? channelWorkerShutdownError;
|
||||
if (finalErr) {
|
||||
if (
|
||||
channelWorkerShutdownError &&
|
||||
channelWorkerManager?.state().enabled
|
||||
) {
|
||||
closePromise = undefined;
|
||||
shuttingDown = false;
|
||||
channelControlDraining = false;
|
||||
}
|
||||
const retryableChannelClose =
|
||||
channelWorkerShutdownError !== undefined &&
|
||||
channelWorkerManager?.state().enabled === true;
|
||||
if (retryableChannelClose) {
|
||||
await flushDaemonLogBounded(
|
||||
daemonLog,
|
||||
DAEMON_LOG_FORCED_FLUSH_BUDGET_MS,
|
||||
);
|
||||
closePromise = undefined;
|
||||
shuttingDown = false;
|
||||
channelControlDraining = false;
|
||||
rej(finalErr);
|
||||
} else {
|
||||
res();
|
||||
return;
|
||||
}
|
||||
if (loggerPublished || loggerSignalOwned) {
|
||||
writeDaemonLifecycleBestEffort(() =>
|
||||
daemonLog.info('daemon stopped'),
|
||||
);
|
||||
await daemonLog.close();
|
||||
}
|
||||
if (finalErr) rej(finalErr);
|
||||
else res();
|
||||
});
|
||||
};
|
||||
|
||||
@@ -5409,23 +5525,33 @@ export async function runQwenServe(
|
||||
}
|
||||
|
||||
if (deps.resolveOnListen) {
|
||||
loggerPublished = true;
|
||||
loggerLifecycle.published();
|
||||
resolve(handle);
|
||||
} else {
|
||||
void runtimeReady.then(
|
||||
() => resolve(handle),
|
||||
() => {
|
||||
loggerPublished = true;
|
||||
loggerLifecycle.published();
|
||||
resolve(handle);
|
||||
},
|
||||
(err) => {
|
||||
void handle.close().then(
|
||||
() => {
|
||||
reject(err instanceof Error ? err : new Error(String(err)));
|
||||
},
|
||||
(closeErr) => {
|
||||
daemonLog.error(
|
||||
'shutdown after runtime startup error failed',
|
||||
closeErr instanceof Error ? closeErr : null,
|
||||
writeDaemonLifecycleBestEffort(() =>
|
||||
daemonLog.error(
|
||||
'shutdown after runtime startup error failed',
|
||||
closeErr instanceof Error ? closeErr : null,
|
||||
),
|
||||
);
|
||||
if (channelWorkerManager?.state().enabled) {
|
||||
daemonLog.error(
|
||||
'runtime startup failed, but qwen serve remains alive to retain the channel service lease until worker exit is confirmed',
|
||||
writeDaemonLifecycleBestEffort(() =>
|
||||
daemonLog.error(
|
||||
'runtime startup failed, but qwen serve remains alive to retain the channel service lease until worker exit is confirmed',
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -166,7 +166,16 @@ function fakeDaemonLog(): DaemonLogger {
|
||||
raw: vi.fn(),
|
||||
getLogPath: () => '',
|
||||
getDaemonId: () => 'test-daemon',
|
||||
getStatus: () => ({
|
||||
runId: '00000000000000000000000000000000',
|
||||
mode: 'stderr-only',
|
||||
health: 'ok',
|
||||
issues: [],
|
||||
droppedRecords: 0,
|
||||
droppedBytes: 0,
|
||||
}),
|
||||
flush: vi.fn(async () => {}),
|
||||
close: vi.fn(async () => {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -22487,37 +22496,41 @@ describe('sendBridgeError daemonLog routing', () => {
|
||||
const tmp = await fsp.mkdtemp(path.join(os.tmpdir(), 'daemon-log-'));
|
||||
const stderrLines: string[] = [];
|
||||
const { initDaemonLogger } = await import('./daemon-logger.js');
|
||||
const daemonLog = initDaemonLogger({
|
||||
const daemonLog = await initDaemonLogger({
|
||||
boundWorkspace: '/w',
|
||||
pid: 1,
|
||||
baseDir: tmp,
|
||||
stderr: (line: string) => stderrLines.push(line),
|
||||
});
|
||||
const bridge = fakeBridge({
|
||||
spawnImpl: async () => {
|
||||
throw new Error('daemon-log-test-boom');
|
||||
},
|
||||
});
|
||||
const app = createServeApp(baseOpts, undefined, { bridge, daemonLog });
|
||||
const res = await request(app)
|
||||
.post('/session')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: '/work/a' });
|
||||
expect(res.status).toBe(500);
|
||||
expect(res.body.error).toBe('daemon-log-test-boom');
|
||||
await daemonLog.flush();
|
||||
// Verify the daemon log file contains the structured error
|
||||
const logPath = daemonLog.getLogPath();
|
||||
const logContent = await fsp.readFile(logPath, 'utf8');
|
||||
expect(logContent).toContain('[ERROR]');
|
||||
expect(logContent).toContain('[DAEMON]');
|
||||
expect(logContent).toContain('daemon-log-test-boom');
|
||||
expect(logContent).toContain('route=POST /session');
|
||||
// Verify stderr also received the line (tee behavior)
|
||||
expect(stderrLines.some((l) => l.includes('daemon-log-test-boom'))).toBe(
|
||||
true,
|
||||
);
|
||||
await fsp.rm(tmp, { recursive: true, force: true });
|
||||
try {
|
||||
const bridge = fakeBridge({
|
||||
spawnImpl: async () => {
|
||||
throw new Error('daemon-log-test-boom');
|
||||
},
|
||||
});
|
||||
const app = createServeApp(baseOpts, undefined, { bridge, daemonLog });
|
||||
const res = await request(app)
|
||||
.post('/session')
|
||||
.set('Host', `127.0.0.1:${baseOpts.port}`)
|
||||
.send({ cwd: '/work/a' });
|
||||
expect(res.status).toBe(500);
|
||||
expect(res.body.error).toBe('daemon-log-test-boom');
|
||||
await daemonLog.flush();
|
||||
// Verify the daemon log file contains the structured error
|
||||
const logPath = daemonLog.getLogPath();
|
||||
const logContent = await fsp.readFile(logPath, 'utf8');
|
||||
expect(logContent).toContain('[ERROR]');
|
||||
expect(logContent).toContain('[DAEMON]');
|
||||
expect(logContent).toContain('daemon-log-test-boom');
|
||||
expect(logContent).toContain('route=POST /session');
|
||||
// Verify stderr also received the line (tee behavior)
|
||||
expect(stderrLines.some((l) => l.includes('daemon-log-test-boom'))).toBe(
|
||||
true,
|
||||
);
|
||||
} finally {
|
||||
await daemonLog.close();
|
||||
await fsp.rm(tmp, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('falls back to writeStderrLine when daemonLog is not provided', async () => {
|
||||
|
||||
@@ -0,0 +1,219 @@
|
||||
/**
|
||||
* @license
|
||||
* Copyright 2026 Qwen Team
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'node:events';
|
||||
import type { Application, RequestHandler } from 'express';
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import type { DaemonLogContext, DaemonLogger } from '../daemon-logger.js';
|
||||
import { installAccessLogMiddleware } from './access-log.js';
|
||||
|
||||
function fakeLogger(): DaemonLogger {
|
||||
return {
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
raw: vi.fn(),
|
||||
getLogPath: () => '',
|
||||
getDaemonId: () => 'daemon:test',
|
||||
getStatus: () => ({
|
||||
runId: '00000000000000000000000000000000',
|
||||
mode: 'stderr-only',
|
||||
health: 'ok',
|
||||
issues: [],
|
||||
droppedRecords: 0,
|
||||
droppedBytes: 0,
|
||||
}),
|
||||
flush: vi.fn(async () => {}),
|
||||
close: vi.fn(async () => {}),
|
||||
};
|
||||
}
|
||||
|
||||
function harness() {
|
||||
let middleware: RequestHandler | undefined;
|
||||
const app = {
|
||||
locals: {},
|
||||
use: vi.fn((handler: RequestHandler) => {
|
||||
middleware = handler;
|
||||
}),
|
||||
} as unknown as Application;
|
||||
const logger = fakeLogger();
|
||||
let now = 1_000;
|
||||
const controller = installAccessLogMiddleware(app, logger, () => now);
|
||||
|
||||
const begin = (
|
||||
input: {
|
||||
path?: string;
|
||||
method?: string;
|
||||
status?: number;
|
||||
rawHeaders?: string[];
|
||||
} = {},
|
||||
) => {
|
||||
if (!middleware) throw new Error('Access middleware was not installed');
|
||||
const response = new EventEmitter() as EventEmitter & {
|
||||
statusCode: number;
|
||||
};
|
||||
response.statusCode = input.status ?? 200;
|
||||
const next = vi.fn();
|
||||
middleware(
|
||||
{
|
||||
method: input.method ?? 'GET',
|
||||
path: input.path ?? '/test',
|
||||
rawHeaders: input.rawHeaders ?? [],
|
||||
} as never,
|
||||
response as never,
|
||||
next,
|
||||
);
|
||||
return { response, next };
|
||||
};
|
||||
|
||||
return {
|
||||
logger,
|
||||
controller,
|
||||
begin,
|
||||
setNow: (value: number) => {
|
||||
now = value;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe('installAccessLogMiddleware', () => {
|
||||
it('preserves exclusions and never changes HTTP admission', () => {
|
||||
const h = harness();
|
||||
for (const request of [
|
||||
{ method: 'GET', path: '/health', status: 200 },
|
||||
{ method: 'POST', path: '/session/s/heartbeat', status: 204 },
|
||||
{ method: 'GET', path: '/session/s/events', status: 200 },
|
||||
]) {
|
||||
const { response, next } = h.begin(request);
|
||||
expect(next).toHaveBeenCalledOnce();
|
||||
response.emit('finish');
|
||||
}
|
||||
expect(h.logger.info).not.toHaveBeenCalled();
|
||||
expect(h.logger.warn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('routes non-error responses to info and errors to warn', () => {
|
||||
const h = harness();
|
||||
h.begin({ path: '/success', status: 399 }).response.emit('finish');
|
||||
h.begin({ path: '/failure', status: 400 }).response.emit('finish');
|
||||
|
||||
expect(h.logger.info).toHaveBeenCalledWith(
|
||||
'request completed',
|
||||
expect.objectContaining({ route: 'GET /success', status: 399 }),
|
||||
);
|
||||
expect(h.logger.warn).toHaveBeenCalledWith(
|
||||
'request completed',
|
||||
expect.objectContaining({ route: 'GET /failure', status: 400 }),
|
||||
);
|
||||
});
|
||||
|
||||
it('caps UTF-8 fields, uses the first raw client header, and tolerates clock retreat', () => {
|
||||
const h = harness();
|
||||
const sessionId = '你'.repeat(100);
|
||||
const routeSuffix = '界'.repeat(900);
|
||||
const firstClientId = '客'.repeat(100);
|
||||
const { response } = h.begin({
|
||||
path: `/session/${sessionId}/${routeSuffix}`,
|
||||
rawHeaders: [
|
||||
'X-Qwen-Client-Id',
|
||||
firstClientId,
|
||||
'x-qwen-client-id',
|
||||
'second-client',
|
||||
],
|
||||
});
|
||||
h.setNow(900);
|
||||
response.emit('finish');
|
||||
|
||||
const context = vi.mocked(h.logger.info).mock.calls[0]?.[1] as
|
||||
| DaemonLogContext
|
||||
| undefined;
|
||||
expect(context).toBeDefined();
|
||||
expect(
|
||||
Buffer.byteLength(String(context?.route), 'utf8'),
|
||||
).toBeLessThanOrEqual(2 * 1024);
|
||||
expect(context?.['routeOriginalBytes']).toBeGreaterThan(2 * 1024);
|
||||
expect(
|
||||
Buffer.byteLength(String(context?.sessionId), 'utf8'),
|
||||
).toBeLessThanOrEqual(256);
|
||||
expect(context?.['sessionIdOriginalBytes']).toBe(300);
|
||||
expect(
|
||||
Buffer.byteLength(String(context?.clientId), 'utf8'),
|
||||
).toBeLessThanOrEqual(256);
|
||||
expect(context?.['clientIdOriginalBytes']).toBe(300);
|
||||
expect(context?.clientId).not.toBe('second-client');
|
||||
expect(context?.['durationMs']).toBe(0);
|
||||
});
|
||||
|
||||
it('bounds bursts, emits summaries before individual records, and keeps fixed status buckets', () => {
|
||||
const h = harness();
|
||||
for (let i = 0; i < 60; i += 1) {
|
||||
h.begin({ path: `/burst/${i}` }).response.emit('finish');
|
||||
}
|
||||
for (const status of [201, 302, 404, 503, 700]) {
|
||||
h.begin({ path: `/suppressed/${status}`, status }).response.emit(
|
||||
'finish',
|
||||
);
|
||||
}
|
||||
expect(h.logger.info).toHaveBeenCalledTimes(60);
|
||||
|
||||
h.setNow(1_500);
|
||||
h.begin({ path: '/summary-priority', status: 204 }).response.emit('finish');
|
||||
expect(h.logger.warn).toHaveBeenCalledWith('access logs suppressed', {
|
||||
suppressed: 5,
|
||||
status2xx: 1,
|
||||
status3xx: 1,
|
||||
status4xx: 1,
|
||||
status5xx: 1,
|
||||
statusOther: 1,
|
||||
});
|
||||
expect(h.logger.info).toHaveBeenCalledTimes(60);
|
||||
|
||||
h.controller.sealAndFlushSuppressed();
|
||||
h.controller.sealAndFlushSuppressed();
|
||||
expect(h.logger.warn).toHaveBeenLastCalledWith('access logs suppressed', {
|
||||
suppressed: 1,
|
||||
status2xx: 1,
|
||||
status3xx: 0,
|
||||
status4xx: 0,
|
||||
status5xx: 0,
|
||||
statusOther: 0,
|
||||
});
|
||||
expect(h.logger.warn).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('does not move the token refill baseline backward with the clock', () => {
|
||||
const h = harness();
|
||||
for (let i = 0; i < 60; i += 1) {
|
||||
h.begin({ path: `/burst/${i}` }).response.emit('finish');
|
||||
}
|
||||
|
||||
h.setNow(0);
|
||||
h.begin({ path: '/clock-retreat' }).response.emit('finish');
|
||||
h.setNow(1_200);
|
||||
h.begin({ path: '/small-forward-step' }).response.emit('finish');
|
||||
|
||||
expect(h.logger.info).toHaveBeenCalledTimes(60);
|
||||
expect(h.logger.warn).not.toHaveBeenCalled();
|
||||
h.controller.sealAndFlushSuppressed();
|
||||
expect(h.logger.warn).toHaveBeenCalledWith('access logs suppressed', {
|
||||
suppressed: 2,
|
||||
status2xx: 2,
|
||||
status3xx: 0,
|
||||
status4xx: 0,
|
||||
status5xx: 0,
|
||||
statusOther: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it('ignores finish callbacks that arrive after sealing', () => {
|
||||
const h = harness();
|
||||
const pending = h.begin({ path: '/late', status: 500 });
|
||||
h.controller.sealAndFlushSuppressed();
|
||||
pending.response.emit('finish');
|
||||
expect(h.logger.info).not.toHaveBeenCalled();
|
||||
expect(h.logger.warn).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -4,16 +4,120 @@
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
|
||||
import { performance } from 'node:perf_hooks';
|
||||
import type { Application } from 'express';
|
||||
import type { DaemonLogger } from '../daemon-logger.js';
|
||||
import type { DaemonLogContext, DaemonLogger } from '../daemon-logger.js';
|
||||
|
||||
const SESSION_ID_RE = /\/session\/([^/]+)/;
|
||||
const ACCESS_LOG_BURST = 60;
|
||||
const ACCESS_LOG_REFILL_PER_SECOND = 2;
|
||||
const ROUTE_MAX_BYTES = 2 * 1024;
|
||||
const SESSION_ID_MAX_BYTES = 256;
|
||||
const CLIENT_ID_MAX_BYTES = 256;
|
||||
|
||||
export const ACCESS_LOG_CONTROLLER_LOCAL = 'accessLogController';
|
||||
|
||||
export interface AccessLogController {
|
||||
sealAndFlushSuppressed(): void;
|
||||
}
|
||||
|
||||
export interface AccessLogAppLocals {
|
||||
[ACCESS_LOG_CONTROLLER_LOCAL]?: AccessLogController;
|
||||
}
|
||||
|
||||
interface SuppressedCounts extends DaemonLogContext {
|
||||
suppressed: number;
|
||||
status2xx: number;
|
||||
status3xx: number;
|
||||
status4xx: number;
|
||||
status5xx: number;
|
||||
statusOther: number;
|
||||
}
|
||||
|
||||
function emptySuppressedCounts(): SuppressedCounts {
|
||||
return {
|
||||
suppressed: 0,
|
||||
status2xx: 0,
|
||||
status3xx: 0,
|
||||
status4xx: 0,
|
||||
status5xx: 0,
|
||||
statusOther: 0,
|
||||
};
|
||||
}
|
||||
|
||||
function countSuppressed(counts: SuppressedCounts, status: number): void {
|
||||
counts.suppressed += 1;
|
||||
if (status >= 200 && status < 300) counts.status2xx += 1;
|
||||
else if (status >= 300 && status < 400) counts.status3xx += 1;
|
||||
else if (status >= 400 && status < 500) counts.status4xx += 1;
|
||||
else if (status >= 500 && status < 600) counts.status5xx += 1;
|
||||
else counts.statusOther += 1;
|
||||
}
|
||||
|
||||
function firstRawHeader(
|
||||
rawHeaders: readonly string[],
|
||||
targetName: string,
|
||||
): string | undefined {
|
||||
for (let i = 0; i + 1 < rawHeaders.length; i += 2) {
|
||||
if (rawHeaders[i]?.toLowerCase() === targetName) return rawHeaders[i + 1];
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function truncateUtf8(
|
||||
value: string,
|
||||
maxBytes: number,
|
||||
): { value: string; originalBytes?: number } {
|
||||
const bytes = Buffer.from(value, 'utf8');
|
||||
if (bytes.length <= maxBytes) return { value };
|
||||
let end = maxBytes;
|
||||
while (end > 0 && (bytes[end] & 0xc0) === 0x80) end -= 1;
|
||||
return {
|
||||
value: bytes.subarray(0, end).toString('utf8'),
|
||||
originalBytes: bytes.length,
|
||||
};
|
||||
}
|
||||
|
||||
export function installAccessLogMiddleware(
|
||||
app: Application,
|
||||
daemonLog: DaemonLogger | undefined,
|
||||
): void {
|
||||
if (!daemonLog) return;
|
||||
monotonicNow: () => number = () => performance.now(),
|
||||
): AccessLogController {
|
||||
let sealed = false;
|
||||
let tokens = ACCESS_LOG_BURST;
|
||||
let refillBaseline = monotonicNow();
|
||||
let suppressed = emptySuppressedCounts();
|
||||
|
||||
const refill = (): void => {
|
||||
const now = Math.max(monotonicNow(), refillBaseline);
|
||||
tokens = Math.min(
|
||||
ACCESS_LOG_BURST,
|
||||
tokens + ((now - refillBaseline) / 1_000) * ACCESS_LOG_REFILL_PER_SECOND,
|
||||
);
|
||||
refillBaseline = now;
|
||||
};
|
||||
|
||||
const flushSuppressed = (): boolean => {
|
||||
if (!daemonLog || suppressed.suppressed === 0) return false;
|
||||
daemonLog.warn('access logs suppressed', suppressed);
|
||||
suppressed = emptySuppressedCounts();
|
||||
return true;
|
||||
};
|
||||
|
||||
const controller: AccessLogController = {
|
||||
sealAndFlushSuppressed: () => {
|
||||
if (sealed) return;
|
||||
sealed = true;
|
||||
try {
|
||||
flushSuppressed();
|
||||
} catch {
|
||||
// Diagnostic logging must not prevent daemon shutdown.
|
||||
}
|
||||
},
|
||||
};
|
||||
(app.locals as AccessLogAppLocals)[ACCESS_LOG_CONTROLLER_LOCAL] = controller;
|
||||
|
||||
if (!daemonLog) return controller;
|
||||
|
||||
app.use((req, res, next) => {
|
||||
const { method, path: reqPath } = req;
|
||||
@@ -23,33 +127,66 @@ export function installAccessLogMiddleware(
|
||||
) {
|
||||
return next();
|
||||
}
|
||||
const startMs = Date.now();
|
||||
const startMs = monotonicNow();
|
||||
res.on('finish', () => {
|
||||
try {
|
||||
if (sealed) return;
|
||||
const status = res.statusCode;
|
||||
if (method === 'GET' && reqPath.endsWith('/events') && status === 200) {
|
||||
return;
|
||||
}
|
||||
const durationMs = Date.now() - startMs;
|
||||
const sessionMatch = reqPath.match(SESSION_ID_RE);
|
||||
const sessionId = sessionMatch?.[1];
|
||||
const clientId = req.headers['x-qwen-client-id'] as string | undefined;
|
||||
const ctx = {
|
||||
route: `${method} ${reqPath}`,
|
||||
...(sessionId ? { sessionId } : {}),
|
||||
...(clientId ? { clientId } : {}),
|
||||
status,
|
||||
durationMs,
|
||||
};
|
||||
if (status >= 400) {
|
||||
daemonLog.warn('request completed', ctx);
|
||||
} else {
|
||||
daemonLog.info('request completed', ctx);
|
||||
refill();
|
||||
if (suppressed.suppressed > 0 && tokens >= 1) {
|
||||
tokens -= 1;
|
||||
flushSuppressed();
|
||||
}
|
||||
if (tokens < 1) {
|
||||
countSuppressed(suppressed, status);
|
||||
return;
|
||||
}
|
||||
tokens -= 1;
|
||||
|
||||
const route = truncateUtf8(`${method} ${reqPath}`, ROUTE_MAX_BYTES);
|
||||
const sessionMatch = reqPath.match(SESSION_ID_RE);
|
||||
const sessionId = sessionMatch?.[1]
|
||||
? truncateUtf8(sessionMatch[1], SESSION_ID_MAX_BYTES)
|
||||
: undefined;
|
||||
const rawClientId = firstRawHeader(req.rawHeaders, 'x-qwen-client-id');
|
||||
const clientId = rawClientId
|
||||
? truncateUtf8(rawClientId, CLIENT_ID_MAX_BYTES)
|
||||
: undefined;
|
||||
const ctx = {
|
||||
route: route.value,
|
||||
...(route.originalBytes
|
||||
? { routeOriginalBytes: route.originalBytes }
|
||||
: {}),
|
||||
...(sessionId
|
||||
? {
|
||||
sessionId: sessionId.value,
|
||||
...(sessionId.originalBytes
|
||||
? { sessionIdOriginalBytes: sessionId.originalBytes }
|
||||
: {}),
|
||||
}
|
||||
: {}),
|
||||
...(clientId
|
||||
? {
|
||||
clientId: clientId.value,
|
||||
...(clientId.originalBytes
|
||||
? { clientIdOriginalBytes: clientId.originalBytes }
|
||||
: {}),
|
||||
}
|
||||
: {}),
|
||||
status,
|
||||
durationMs: Math.max(0, Math.round(monotonicNow() - startMs)),
|
||||
};
|
||||
if (status >= 400) daemonLog.warn('request completed', ctx);
|
||||
else daemonLog.info('request completed', ctx);
|
||||
} catch {
|
||||
// Logging failure must not affect the request.
|
||||
}
|
||||
});
|
||||
next();
|
||||
});
|
||||
|
||||
return controller;
|
||||
}
|
||||
@@ -493,6 +493,9 @@ export type {
|
||||
DaemonPreflightKind,
|
||||
DaemonStatus,
|
||||
DaemonStatusCell,
|
||||
DaemonLogHealth,
|
||||
DaemonLogIssue,
|
||||
DaemonLogMode,
|
||||
DaemonStatusReport,
|
||||
DaemonStatusReportDetail,
|
||||
DaemonStatusReportIssue,
|
||||
|
||||
@@ -275,6 +275,16 @@ export type DaemonStatusReportDetail = 'summary' | 'full';
|
||||
/** Overall health rollup of a daemon status report. */
|
||||
export type DaemonStatusReportLevel = 'ok' | 'warning' | 'error';
|
||||
|
||||
export type DaemonLogMode = 'stable' | 'fallback' | 'stderr-only';
|
||||
export type DaemonLogHealth = 'ok' | 'degraded';
|
||||
export type DaemonLogIssue =
|
||||
| 'init_failed'
|
||||
| 'rotation_failed'
|
||||
| 'retention_failed'
|
||||
| 'queue_overflow'
|
||||
| 'write_failed'
|
||||
| 'lease_compromised';
|
||||
|
||||
/** One triage finding surfaced by the daemon status rollup. */
|
||||
export interface DaemonStatusReportIssue {
|
||||
code: string;
|
||||
@@ -417,8 +427,17 @@ export interface DaemonStatusReport {
|
||||
};
|
||||
qwenCodeVersion?: string;
|
||||
daemonId?: string;
|
||||
runId?: string;
|
||||
logMode?: DaemonLogMode;
|
||||
logHealth?: DaemonLogHealth;
|
||||
/** Present only in `detail=full` responses. */
|
||||
logPath?: string;
|
||||
/** Present only in `detail=full` responses. */
|
||||
logIssues?: readonly DaemonLogIssue[];
|
||||
/** Present only in `detail=full` responses. */
|
||||
logDroppedRecords?: number;
|
||||
/** Present only in `detail=full` responses. */
|
||||
logDroppedBytes?: number;
|
||||
};
|
||||
security: {
|
||||
tokenConfigured: boolean;
|
||||
|
||||
@@ -192,6 +192,9 @@ export {
|
||||
type DaemonChannelWorkerGroupSnapshot,
|
||||
type DaemonChannelWorkerSnapshot,
|
||||
type DaemonStatusCell,
|
||||
type DaemonLogHealth,
|
||||
type DaemonLogIssue,
|
||||
type DaemonLogMode,
|
||||
type DaemonStatusReport,
|
||||
type DaemonStatusReportDetail,
|
||||
type DaemonStatusReportIssue,
|
||||
|
||||
@@ -73,6 +73,9 @@ import type {
|
||||
DaemonSessionUpdateData,
|
||||
DaemonSessionUpdateEvent,
|
||||
DaemonSessionViewState,
|
||||
DaemonLogHealth,
|
||||
DaemonLogIssue,
|
||||
DaemonLogMode,
|
||||
DaemonStatusReport,
|
||||
DaemonStatusReportDetail,
|
||||
DaemonStatusReportIssue,
|
||||
@@ -268,9 +271,20 @@ describe('public SDK entry — typed daemon event surface (#4217)', () => {
|
||||
// `GET /daemon/status` report surface (PR 5174 client coverage): the
|
||||
// envelope plus the sub-shapes UI dashboards need to type against.
|
||||
expectTypeOf<DaemonStatusReport>().not.toBeNever();
|
||||
expectTypeOf<DaemonLogMode>().not.toBeNever();
|
||||
expectTypeOf<DaemonLogHealth>().not.toBeNever();
|
||||
expectTypeOf<DaemonLogIssue>().not.toBeNever();
|
||||
expectTypeOf<DaemonStatusReport['limits']>().toMatchTypeOf<{
|
||||
compactedReplayMaxBytes: number;
|
||||
}>();
|
||||
expectTypeOf<DaemonStatusReport['daemon']>().toMatchTypeOf<{
|
||||
runId?: string;
|
||||
logMode?: DaemonLogMode;
|
||||
logHealth?: DaemonLogHealth;
|
||||
logIssues?: readonly DaemonLogIssue[];
|
||||
logDroppedRecords?: number;
|
||||
logDroppedBytes?: number;
|
||||
}>();
|
||||
expectTypeOf<DaemonStatusReportDetail>().not.toBeNever();
|
||||
expectTypeOf<DaemonStatusReportIssue>().not.toBeNever();
|
||||
expectTypeOf<DaemonStatusReportLevel>().not.toBeNever();
|
||||
|
||||
Reference in new issue
Block a user