20260914 完善OSS日志与daemon部署

docker/compose.all5-daemon.yaml

@@ -0,0 +1,73 @@
+services:
+  validator:
+    build:
+      context: ..
+      dockerfile: docker/Dockerfile
+    image: "${PANDA_RPKI_IMAGE:-panda-rpki:v0.1.0}"
+    user: "${PUID:-1000}:${PGID:-1000}"
+    read_only: true
+    init: true
+    restart: unless-stopped
+    stop_grace_period: 40s
+    tmpfs:
+      - /tmp
+    volumes:
+      - ${INPUT_DIR:?set INPUT_DIR to the TAL/TA directory}:/input:ro
+      - ${DAEMON_DIR:?set DAEMON_DIR to a writable dedicated directory}:/data
+    environment:
+      PANDA_RPKI_LOG_LEVEL: "${PANDA_RPKI_LOG_LEVEL:-info}"
+      PANDA_RPKI_LOG_FORMAT: "${PANDA_RPKI_LOG_FORMAT:-text}"
+      PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS: "${PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS:-10}"
+    command:
+      - daemon
+      - --state-root
+      - /data
+      - --interval-secs
+      - "${PANDA_RPKI_INTERVAL_SECS:-60}"
+      - --retain-runs
+      - "${PANDA_RPKI_RETAIN_RUNS:-10}"
+      - --run-timeout-secs
+      - "${PANDA_RPKI_RUN_TIMEOUT_SECS:-0}"
+      - --shutdown-grace-secs
+      - "${PANDA_RPKI_SHUTDOWN_GRACE_SECS:-30}"
+      - --
+      - --tal
+      - /input/${AFRINIC_TAL_FILE:?set AFRINIC_TAL_FILE}
+      - --ta
+      - /input/${AFRINIC_TA_FILE:?set AFRINIC_TA_FILE}
+      - --tal-id
+      - afrinic
+      - --tal
+      - /input/${APNIC_TAL_FILE:?set APNIC_TAL_FILE}
+      - --ta
+      - /input/${APNIC_TA_FILE:?set APNIC_TA_FILE}
+      - --tal-id
+      - apnic
+      - --tal
+      - /input/${ARIN_TAL_FILE:?set ARIN_TAL_FILE}
+      - --ta
+      - /input/${ARIN_TA_FILE:?set ARIN_TA_FILE}
+      - --tal-id
+      - arin
+      - --tal
+      - /input/${LACNIC_TAL_FILE:?set LACNIC_TAL_FILE}
+      - --ta
+      - /input/${LACNIC_TA_FILE:?set LACNIC_TA_FILE}
+      - --tal-id
+      - lacnic
+      - --tal
+      - /input/${RIPE_TAL_FILE:?set RIPE_TAL_FILE}
+      - --ta
+      - /input/${RIPE_TA_FILE:?set RIPE_TA_FILE}
+      - --tal-id
+      - ripe
+      - --parallel-phase2-object-workers
+      - "${PANDA_RPKI_WORKERS:-8}"
+      - --parallel-max-repo-sync-workers-global
+      - "${PANDA_RPKI_REPO_WORKERS:-8}"
+      - --parallel-phase2-worker-queue-capacity
+      - "${PANDA_RPKI_WORKER_QUEUE_CAPACITY:-256}"
+      - --parallel-repo-worker-queue-capacity
+      - "${PANDA_RPKI_REPO_QUEUE_CAPACITY:-256}"
+      - --http-timeout-secs
+      - "${PANDA_RPKI_HTTP_TIMEOUT_SECS:-600}"

docker/compose.daemon.yaml

@@ -16,7 +16,7 @@ services:
       - ${DAEMON_DIR:?set DAEMON_DIR to a writable dedicated directory}:/data
     environment:
       PANDA_RPKI_LOG_LEVEL: "${PANDA_RPKI_LOG_LEVEL:-info}"
-      PANDA_RPKI_LOG_FORMAT: "${PANDA_RPKI_LOG_FORMAT:-json}"
+      PANDA_RPKI_LOG_FORMAT: "${PANDA_RPKI_LOG_FORMAT:-text}"
     command:
       - daemon
       - --state-root

docs/command-line-reference.md

@@ -73,6 +73,7 @@ can increase memory consumption and does not guarantee proportional speedup.
 | Option | Default | Values |
 | --- | --- | --- |
 | `--log-level <level>` | `info` | `off`, `error`, `warn`, `info`, `debug`, `trace` |
+| `--log-progress-interval-secs <seconds>` | `10` | Periodic progress; `0` disables periodic events, not stage events. CLI overrides `PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS`. |
 | `--log-format <format>` | `text` | `text`, `json` |
 
 Precedence: explicit option, then `PANDA_RPKI_LOG_LEVEL` or
@@ -125,7 +126,7 @@ has stopped before restarting. Stop the entire container for Docker deployments.
 Daemon adds `daemon.lock`, `daemon-status.json`, `lifecycle.json`,
 `run-summary.jsonl`, `state/repository-db/`, and `runs/run_000001/` etc.
 Each run includes validator outputs, `run-meta.json`, `run-summary.json`,
-`stdout.log`, `stderr.log` and optional `process-time.txt`. `/usr/bin/time`
+`process-output.txt`, `validator.log` and optional `process-time.txt`. `/usr/bin/time`
 provides user/system CPU and peak RSS when installed; Docker includes it.
 Controller states: `starting`, `running`, `sleeping`, `exited`, `failed`.
 

docs/docker.md

@@ -47,6 +47,7 @@ docker compose -f docker/compose.all5.yaml run --build --rm validator
 | `PANDA_RPKI_WORKERS` | 8 | all5 and daemon object workers |
 | `PANDA_RPKI_REPO_WORKERS` | 8 | all5 and daemon transport workers |
 | `PANDA_RPKI_WORKER_QUEUE_CAPACITY` | 256 | all5 object queue |
+| `PANDA_RPKI_REPO_QUEUE_CAPACITY` | 256 | all5 daemon repository queue |
 | `PANDA_RPKI_HTTP_TIMEOUT_SECS` | 600 | all5 HTTP timeout (native CLI default is 300) |
 | `PANDA_RPKI_RRDP_SYNC_MODE` | auto | single-anchor and all5 |
 
@@ -54,7 +55,121 @@ The all5 CCR is `OUTPUT_DIR/all5.ccr`; the single-anchor CCR is
 `OUTPUT_DIR/result.ccr`. Reuse `STATE_DIR` and select new `OUTPUT_DIR` values
 to retain multiple cycles. These templates take no bundled RIR input files.
 
-## Daemon service
+## All-five daemon service
+
+The all-five daemon template combines the five regional TAL/TA pairs with the
+continuous daemon controller. Prepare the same `input/` directory used by the
+all-five one-shot example, then run:
+
+```bash
+mkdir -p data/all5-daemon
+export INPUT_DIR="$PWD/input" DAEMON_DIR="$PWD/data/all5-daemon"
+export AFRINIC_TAL_FILE=afrinic.tal AFRINIC_TA_FILE=afrinic.cer
+export APNIC_TAL_FILE=apnic.tal APNIC_TA_FILE=apnic.cer
+export ARIN_TAL_FILE=arin.tal ARIN_TA_FILE=arin.cer
+export LACNIC_TAL_FILE=lacnic.tal LACNIC_TA_FILE=lacnic.cer
+export RIPE_TAL_FILE=ripe.tal RIPE_TA_FILE=ripe.cer
+export PUID="$(id -u)" PGID="$(id -g)"
+docker compose -f docker/compose.all5-daemon.yaml up -d --build
+docker compose -f docker/compose.all5-daemon.yaml logs -f validator
+```
+
+The template uses a 60-second interval by default, keeps 10 completed runs,
+and leaves the daemon running until it is stopped. It writes each run below
+`DAEMON_DIR/runs/` and keeps shared protocol state below `DAEMON_DIR/state/`.
+The default diagnostic format is `text`, with a source file and line appended
+to each diagnostic record; select `json` explicitly for machine consumers.
+Set `PANDA_RPKI_INTERVAL_SECS` only when a different interval is intended. Stop
+the service without deleting its data with:
+
+```bash
+docker compose -f docker/compose.all5-daemon.yaml stop
+```
+
+## Upgrade a daemon deployment
+
+An upgrade replaces the image and recreates the container. It does not replace
+the host data directory. Keep the same `DAEMON_DIR`, Compose project, TAL/TA
+inputs and controller settings so the new image can resume the existing
+protocol state. Do not run two writable daemons against the same state root.
+
+Run the following from the deployment directory that owns the Compose file. Use
+a unique image tag for every candidate so the image used by a run is auditable:
+
+```bash
+export COMPOSE_FILE="${COMPOSE_FILE:-docker/compose.all5-daemon.yaml}"
+export COMPOSE_PROJECT_NAME=panda-rpki-all5
+export PANDA_RPKI_IMAGE=panda-rpki:2026-09-11-logging
+export SOURCE_DIR=/path/to/panda-rpki-oss
+
+# Keep the existing INPUT_DIR, DAEMON_DIR, TAL/TA variables and settings.
+docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" ps
+test "$(jq -r .state "$DAEMON_DIR/daemon-status.json")" = sleeping
+
+# Build from the source tree before stopping the running service.
+docker build -f "$SOURCE_DIR/docker/Dockerfile" -t "$PANDA_RPKI_IMAGE" "$SOURCE_DIR"
+docker image inspect "$PANDA_RPKI_IMAGE" --format '{{.Id}}'
+
+# Save configuration and create a consistent stopped-state backup.
+backup="upgrade-backup-$(date -u +%Y%m%dT%H%M%SZ)"
+mkdir -p "$backup"
+cp "$COMPOSE_FILE" "$backup/compose.yaml"
+if test -f .env; then cp .env "$backup/env"; fi
+docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" stop
+cp -a "$DAEMON_DIR/state" "$backup/state"
+cp -a "$DAEMON_DIR/daemon-status.json" "$DAEMON_DIR/lifecycle.json" "$backup/"
+
+# Recreate the same service with the new image; bind-mounted data is retained.
+docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" up -d --no-build --force-recreate
+docker inspect "$(docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" ps -q validator)" \
+  --format 'image={{.Config.Image}} running={{.State.Running}}'
+```
+
+After the first upgraded run, verify that the run sequence advanced, the run
+completed, and the new log/outputs exist. The exact run directory is printed by
+the daemon status and can be selected with `find`:
+
+```bash
+latest_run="$(find "$DAEMON_DIR/runs" -mindepth 1 -maxdepth 1 -type d -name 'run_*' | sort | tail -1)"
+test -s "$latest_run/run-summary.json"
+test -s "$latest_run/validator.log"
+test -s "$latest_run/result.ccr"
+grep -Eq '\[src/[^]]+\.rs:[0-9]+\]$' "$latest_run/validator.log"
+jq . "$latest_run/run-summary.json"
+```
+
+The upgrade process preserves completed run directories until normal retention
+removes them. Archive important runs separately if they must outlive
+`PANDA_RPKI_RETAIN_RUNS`.
+
+### Roll back
+
+Stop the upgraded service before changing the image. Restore the previous
+Compose file and `.env`, or set `PANDA_RPKI_IMAGE` to the previously verified
+tag, then recreate the service with the same project and data directory:
+
+```bash
+docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" stop
+cp "$backup/compose.yaml" "$COMPOSE_FILE"
+if test -f "$backup/env"; then cp "$backup/env" .env; fi
+unset PANDA_RPKI_IMAGE
+docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" up -d --no-build --force-recreate
+```
+
+If the new binary changed the state schema or wrote incompatible state, stop the
+service and move the current state aside before restoring the matching
+stopped-state backup:
+
+```bash
+mv "$DAEMON_DIR/state" "$DAEMON_DIR/state-incompatible-$(date -u +%Y%m%dT%H%M%SZ)"
+cp -a "$backup/state" "$DAEMON_DIR/state"
+```
+
+Never delete `DAEMON_DIR` or use `docker compose down -v` as part of a routine
+upgrade. Check [state compatibility](command-line-reference.md#state-and-network)
+before deploying a binary with storage changes.
+
+## Single-anchor daemon service
 
 ```bash
 mkdir -p data/daemon
@@ -69,7 +184,7 @@ docker compose -f docker/compose.daemon.yaml stop
 ```
 
 Defaults: interval 60 seconds, retain 10, `PANDA_RPKI_RUN_TIMEOUT_SECS=0`
-(disabled), log level `info`, log format `json`. The daemon template passes
+(disabled), log level `info`, log format `text`. The daemon template passes
 `PANDA_RPKI_LOG_LEVEL` and `PANDA_RPKI_LOG_FORMAT` into the container.
 It restarts unless stopped and allows 40 seconds for shutdown, exceeding the
 controller's default 30-second grace plus termination wait.

docs/logging.md

@@ -0,0 +1,124 @@
+# Diagnostic logging
+
+The validator writes diagnostics to stderr. Under daemon, that stream is saved
+as `runs/run_XXXXXX/validator.log`; it contains all enabled levels, not only errors.
+`process-output.txt` captures ordinary output and is normally empty. The daemon's
+own lifecycle diagnostics remain in the container log. No duplicate log forwarding
+is performed. Historical `stderr.log` files are not renamed.
+
+CLI and both daemon Compose files default to `info` and `text`.
+Text is written for people, with independently authored English sentences for
+run lifecycle, validation stages, progress, output files and failures. It does not
+enumerate event fields or translate event identifiers into prose. Progress reports
+elapsed time, publication points processed when known, and active repository sync
+attempts. Old progress snapshots are explicitly described as the last reported
+update. UUIDs, worker queues and unrelated counters are not printed.
+The synthetic initializing-stage completion is omitted in text only. Control
+characters are escaped to keep each record on one line. Diagnostic filenames do
+not select the format. Text is not a stable machine-parsing interface.
+
+```text
+2026-09-11T10:44:14Z INFO [run 230] Starting validation with delta synchronization across 5 trust anchors.
+2026-09-11T10:44:24Z INFO [run 230] Synchronizing repositories and validating objects; 10.0 seconds elapsed. Processed 32000 publication points so far. 3 repository sync attempts are still in progress.
+2026-09-11T10:45:36Z INFO [run 230] Validation completed in 82.0 seconds.
+```
+
+Use `--log-format json` or `PANDA_RPKI_LOG_FORMAT=json` for machine consumers;
+the JSON event names and fields are unchanged. Both formats redact sensitive data.
+Text error reasons are limited to 240 Unicode characters with an explicit notice
+when shortened. Unknown extension/debug events retain a diagnostic identifier and
+their message, not an arbitrary field dump. Use JSON (and debug/trace as needed)
+for full technical details, category examples and counter snapshots.
+JSON is one complete object per line: `timestamp`, `level`, `event`, `fields`, `source`.
+Timestamps are UTC RFC3339. Run context is in `fields.run_id` and `fields.run_seq`
+(null for standalone validation). Logging failures do not change validation results.
+
+Every emitted text line ends with `[source/file.rs:line]`. JSON records include
+`source: {"file": "source/file.rs", "line": 123}` at the top level. These are
+compile-time source paths and one-based line numbers from the logger call site,
+not the shared output sink; progress forwarding preserves the caller location.
+Logs emitted inside artifact/stage helpers point to the emit call inside that
+helper. Match locations against the source revision used to build the binary.
+
+```text
+2026-09-11T10:45:36Z INFO [run 230] Validation completed in 82.0 seconds. [src/cli/validate.rs:123]
+```
+
+The source line in this example is illustrative; actual logs report the compiled location.
+
+## Event contract
+
+| Events | Level | Meaning / frequency |
+| --- | --- | --- |
+| validation_started/completed/summary | info | Run boundaries and exact final business counts |
+| validation_failed, command_failed | error | Run/command failure, with error and elapsed time where available |
+| validation_stage_started/completed/failed | info/error | State store, setup, TA discovery, online validation, output, CSV conversion |
+| validation_progress | info | Default every 10s, independent of blocked network/store calls |
+| trust_anchor_started/completed/failed | info/error | One pair per TAL, fatal failure includes TAL ID |
+| validation_warnings | warn | Existing final total, not a declaration that the run failed |
+| validation_warning_category | info | Category count and up to three redacted 1KiB examples |
+| validation_warning_detail | debug | Full warning message and context |
+| artifact_write_started/completed/failed | info/error | Type, path, wall milliseconds and bytes; CCR, CSV, summary, timings |
+| rrdp_sync_attempt/success | debug | Transport attempt and successful object writes/duration |
+| rrdp_sync_failed, http_fetch_failed | warn | Recoverable attempt failures; retryable does not imply will_retry |
+| rrdp_sync_selected | debug | Actual snapshot/delta/unchanged selection, reason, serial/session |
+| rrdp_download_completed | debug | Snapshot/delta decoded payload bytes, not wire traffic |
+| rrdp_cross_origin_rejected, rrdp_delta_fallback | warn | Protocol security rejection / delta application fallback |
+| repo_terminal_failure | warn | Failed publication point, not a fatal command error |
+| http_fetch_slow, rsync_sync_slow, phase2_*_slow | debug | Detailed slow operations, aggregated into info progress |
+| phase1_repo_* | debug | Repository task scheduling and worker decisions |
+| phase2 batch/queue/object events, publication_point_* | trace | Detailed high-volume processing; not recommended for continuous all5 |
+| daemon_started/state_recovered/run_started/child_started | info | Effective configuration, recovery and per-run log path |
+| daemon_run_completed/sleeping/retention_applied/exited | info/warn | Outcome, next scheduled time and retention decisions |
+| daemon_stop_requested | info | Graceful stop request observed outside the signal handler |
+| daemon_child_terminate/child_kill | warn | One event per signal escalation, with reason |
+
+`src/logging/progress.rs` explicitly registers progress event levels, including
+the dynamic finalize-worker label; a source-inventory regression detects new
+unregistered literal call sites. Unknown library-supplied names default to debug.
+
+## Progress and volume
+
+`--log-progress-interval-secs` overrides `PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS`,
+default `10`. `0` suppresses periodic events, not stage events. `off/error/warn`
+do not create a progress thread. Queue snapshots are refreshed at most every 250ms.
+`statistics_age_ms` is the age of the queue/PP snapshot, not the repository event
+counters. `progress_stalled` means no observed completion for 30s, not proof of deadlock.
+Progress does not predict the total tree size or invent completion percentages.
+
+- Repository counters count attempts, not publication points or unique repositories.
+- `repository_objects_written` counts writes, including updates; it is not unique validated objects.
+- `roa_tasks_completed` counts returned parallel ROA tasks, not all object types.
+- `downloaded_snapshot_delta_bytes` excludes notifications and measures decoded payload.
+- The serial path reports PP completed/failed counts; absent parallel queue fields are not zeroes.
+- Slow-operation counts are cumulative per run. Detailed slow events stay debug.
+
+Warning categories are assigned at typed source sites. Opaque legacy string
+errors remain `unclassified`; they are not classified by substring guesses.
+The category total equals the existing warning total, which may contain both tree
+and audit warnings. Examples are redacted before UTF-8-safe truncation.
+
+| Category | Assignment at currently instrumented sites | Meaning / example |
+| --- | --- | --- |
+| unclassified | No typed classification retained | Opaque repository-sync error; do not guess its cause |
+| ber_compatible_cms_encoding | Existing BER-compatible CMS warning | Accepted non-DER encoding |
+| transport | Typed fetch error in RRDP fallback warning | Download failure |
+| rrdp_protocol | Typed RRDP error in fallback warning | Invalid RRDP response |
+| security_policy | Cross-origin fallback, manifest location/hash/rollback rejection | Integrity or anti-rollback check |
+| certificate | Child discovery or typed manifest EE/issuer certificate errors | Certificate/path/resource failure |
+| manifest_crl | Typed manifest errors or missing required CRL | Stale manifest or missing CRL |
+| object_validation | Parallel ROA/ASPA rejection | Invalid signed routing object |
+
+This is incremental typed coverage, not a promise that every historical warning
+site is already classified. Uninstrumented sites intentionally remain unclassified.
+
+```bash
+tail -f /path/to/data/runs/run_000001/validator.log
+jq -c 'select(.event == "validation_progress")' /path/to/validator.log
+jq -c 'select(.level == "warn" or .level == "error")' /path/to/validator.log
+```
+
+The `jq` examples require JSON logging to have been explicitly enabled.
+
+Retention removes completed run directories including their logs. Archive needed
+runs before the configured retention window expires. Summary history is append-only.

docs/usage.md

@@ -76,6 +76,15 @@ The append-only summary history is not trimmed by retention.
 
 ## Outputs and diagnostics
 
+`validator.log` is the per-run diagnostic log, not an error-only file.
+`process-output.txt` captures ordinary child output and is normally empty.
+Old runs keep their historical filenames; collectors support `stderr.log` as a fallback.
+Every ten seconds, info logs report the current stage and available progress counters.
+Use `--log-progress-interval-secs 0` to disable periodic events; stage events remain.
+`statistics_age_ms` identifies stale statistics; `progress_stalled` means no observed
+completion for thirty seconds, not a proven deadlock. Totals and percentages are not guessed.
+Events include `run_id` and, under daemon, `run_seq`. Controller logs remain separate.
+
 `vrps.csv` contains `asn,prefix,max_length`. CCR contains canonical state;
 duplicate routing payloads can collapse into a single CCR entry. Count raw
 rows and normalized unique payloads separately when comparing outputs.
@@ -83,7 +92,7 @@ ASPA canonicalization groups providers by customer ASN.
 
 Use `--log-level debug` for diagnosis, or `trace` for detailed object/worker
 events. Logs are written to stderr and can be redirected independently from
-outputs. In daemon mode, inspect per-run `stderr.log` and the controller's
+outputs. In daemon mode, inspect per-run `validator.log` and the controller's
 `daemon-status.json` and `run-summary.jsonl`.
 
 | Symptom | Action |

src/cli/mod.rs

@@ -34,7 +34,10 @@ pub fn usage() -> String {
         "panda-rpki {}\n\nUSAGE:\n  panda-rpki validate --tal <file> --ta <file> [--tal <file> --ta <file> ...] --out <dir> [--rrdp-state-dir <dir>] [--rrdp-sync-mode auto|snapshot|delta] [--ccr-out <file>] [--parallel-phase2-object-workers <n>] [--parallel-max-repo-sync-workers-global <n>] [--parallel-phase2-worker-queue-capacity <n>] [--parallel-repo-worker-queue-capacity <n>] [--max-ca-depth <n>] [--http-timeout-secs <n>] [--http-root-cert <pem>]... [--log-level off|error|warn|info|debug|trace] [--log-format text|json]\n\nTAL/TA pairs are positional and may be repeated. RRDP state is an explicit protocol-resume directory, not a validation cache. The public v0.1.0 fast path accepts HTTPS RRDP notification/snapshot/delta URLs only.",
         crate::VERSION
     );
-    format!("{validation}\n\n{}", crate::daemon::usage())
+    format!(
+        "{validation}\n  --log-progress-interval-secs <n>: periodic info progress (default 10; 0 disables).\n\n{}",
+        crate::daemon::usage()
+    )
 }
 
 pub fn run<I, S>(args: I) -> Result<(), String>

src/cli/validate.rs

@@ -15,6 +15,7 @@ pub(crate) const VALIDATE_OPTIONS: &[&str] = &[
     "tal",
     "log-level",
     "log-format",
+    "log-progress-interval-secs",
     "ta",
     "tal-id",
     "ta-constraints",
@@ -46,6 +47,21 @@ pub(super) fn run(args: &[String]) -> Result<(), String> {
         "text",
     );
     crate::logging::configure(&level, &format)?;
+    let progress_interval = resolve_setting(
+        options.values("log-progress-interval-secs"),
+        std::env::var("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS").ok(),
+        "10",
+    )
+    .parse::<u64>()
+    .map_err(|_| {
+        "invalid log-progress-interval-secs: expected a nonnegative integer".to_string()
+    })?;
+    let run_id =
+        std::env::var("PANDA_RPKI_RUN_ID").unwrap_or_else(|_| uuid::Uuid::new_v4().to_string());
+    let run_seq = std::env::var("PANDA_RPKI_RUN_SEQ")
+        .ok()
+        .and_then(|value| value.parse().ok());
+    crate::logging::set_context(&run_id, run_seq);
     let tal_paths = options.values("tal");
     let ta_paths = options.values("ta");
     if tal_paths.is_empty() || ta_paths.is_empty() || tal_paths.len() != ta_paths.len() {
@@ -156,22 +172,30 @@ pub(super) fn run(args: &[String]) -> Result<(), String> {
     };
 
     let started = std::time::Instant::now();
+    let observer = crate::logging::observer::Observer::start(progress_interval)?;
     crate::logging::emit(crate::logging::Level::Info, "validation_started", || {
         serde_json::json!({
             "tal_count": tal_paths.len(), "tal_ids": tal_ids, "phase2_workers": workers, "repo_workers": repo_workers,
             "sync_mode": format!("{sync_mode:?}").to_lowercase(),
         })
     });
-    crate::runtime::run_config(config).inspect_err(|_| {
+    let result = crate::runtime::run_config(config).and_then(|()| {
+        crate::logging::observer::phase("vrp_csv_conversion");
+        let fast_vrps = out.join(".vrps-source.csv");
+        crate::logging::artifact::write("vrps", &out.join("vrps.csv"), || {
+            convert_vrp_csv(&fast_vrps, &out.join("vrps.csv"))
+        })
+    });
+    observer.finish(result.is_ok());
+    result.inspect_err(|error| {
         crate::logging::emit(crate::logging::Level::Error, "validation_failed", || {
             serde_json::json!({
                 "elapsed_ms": started.elapsed().as_millis(),
+                "error": error,
             })
         });
     })?;
 
-    let fast_vrps = out.join(".vrps-source.csv");
-    convert_vrp_csv(&fast_vrps, &out.join("vrps.csv"))?;
     crate::logging::emit(crate::logging::Level::Info, "validation_completed", || {
         serde_json::json!({
             "elapsed_ms": started.elapsed().as_millis(), "phase2_workers": workers, "repo_workers": repo_workers,

src/daemon/mod.rs

@@ -219,7 +219,19 @@ pub(crate) fn run(args: &[String]) -> Result<(), String> {
         return Err(crate::cli::usage());
     }
     let mut config = parse(args)?;
+    let progress_interval =
+        std::env::var("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS").unwrap_or_else(|_| "10".into());
+    Options::parse(&config.child_args, VALIDATE_OPTIONS, VALIDATE_REPEATABLE)?
+        .values("log-progress-interval-secs")
+        .first()
+        .unwrap_or(&progress_interval)
+        .parse::<u64>()
+        .map_err(|_| "invalid log-progress-interval-secs".to_string())?;
     crate::logging::configure(&config.log_level, &config.log_format)?;
+    emit(Level::Info, "daemon_started", || {
+        json!({"interval_secs": config.interval,
+        "retain_runs": config.retain, "run_timeout_secs": config.timeout, "state_root": config.root})
+    });
     regular_directory(&config.root)?;
     config.root = fs::canonicalize(&config.root).map_err(|e| e.to_string())?;
     let lock = OpenOptions::new()
@@ -269,8 +281,18 @@ fn controller(config: &Config) -> Result<(), String> {
     let mut completed = 0u64;
     let mut failures = 0u64;
     status(root, "starting", lifecycle.last_seq, completed)?;
+    emit(
+        Level::Info,
+        "daemon_state_recovered",
+        || json!({"last_run_seq": lifecycle.last_seq, "last_success": lifecycle.last_success}),
+    );
     while !process::stopping() && config.max_runs.is_none_or(|n| completed < n) {
         let mode = sync_mode(&state, lifecycle.last_success)?;
+        let mode_reason = match mode {
+            "snapshot" => "empty_state",
+            "delta" => "previous_run_succeeded",
+            _ => "previous_run_incomplete_or_failed",
+        };
         lifecycle.last_seq = lifecycle
             .last_seq
             .checked_add(1)
@@ -278,18 +300,20 @@ fn controller(config: &Config) -> Result<(), String> {
         lifecycle.last_success = false;
         write_json_pretty(&lifecycle_path, &lifecycle)?;
         let seq = lifecycle.last_seq;
+        let run_id = uuid::Uuid::new_v4().to_string();
+        crate::logging::set_context(&run_id, Some(seq));
         let out = runs.join(format!("run_{seq:06}"));
         fs::create_dir(&out).map_err(|e| format!("create run: {e}"))?;
         let started_at = now();
         let start = Instant::now();
         let mut meta = json!({"controller": "panda-rpki-daemon-v1", "run_seq": seq,
-            "status": "running", "sync_mode": mode, "started_at": started_at});
+            "status": "running", "sync_mode": mode, "started_at": started_at, "run_id": run_id, "sync_mode_reason": mode_reason});
         write_json_pretty(&out.join("run-meta.json"), &meta)?;
         status(root, "running", seq, completed)?;
         emit(
             Level::Info,
             "daemon_run_started",
-            || json!({"run_seq": seq, "sync_mode": mode}),
+            || json!({"run_seq": seq, "sync_mode": mode, "reason": mode_reason, "validator_log": out.join("validator.log")}),
         );
         // Child-per-run execution and stdout/stderr redirection; paths
         // are supplied as arguments, never rendered into shell command text.
@@ -307,6 +331,9 @@ fn controller(config: &Config) -> Result<(), String> {
             Command::new(&executable)
         };
         command.env("LC_ALL", "C");
+        command
+            .env("PANDA_RPKI_RUN_ID", &run_id)
+            .env("PANDA_RPKI_RUN_SEQ", seq.to_string());
         command
             .arg("validate")
             .args(&config.child_args)
@@ -319,8 +346,8 @@ fn controller(config: &Config) -> Result<(), String> {
             .arg("--rrdp-sync-mode")
             .arg(mode)
             .stdin(Stdio::null())
-            .stdout(File::create(out.join("stdout.log")).map_err(|e| e.to_string())?)
+            .stdout(File::create(out.join("process-output.txt")).map_err(|e| e.to_string())?)
-            .stderr(File::create(out.join("stderr.log")).map_err(|e| e.to_string())?);
+            .stderr(File::create(out.join("validator.log")).map_err(|e| e.to_string())?);
         let result = process::execute(&mut command, config.timeout, config.grace);
         let (exit_code, failure) = match result {
             Ok((exit, reason)) => (
@@ -351,18 +378,34 @@ fn controller(config: &Config) -> Result<(), String> {
         completed += 1;
         failures += u64::from(!success);
         let removed = apply_retention(&runs, config.retain)?;
+        if !removed.is_empty() {
+            emit(
+                Level::Info,
+                "daemon_retention_applied",
+                || json!({"removed_runs": removed, "retain_runs": config.retain}),
+            );
+        }
         emit(
             if success { Level::Info } else { Level::Warn },
             "daemon_run_completed",
-            || json!({"run_seq": seq, "success": success, "wall_ms": meta["wall_ms"], "removed_runs": removed}),
+            || json!({"run_seq": seq, "success": success, "wall_ms": meta["wall_ms"], "removed_runs": removed, "error": meta["error"], "exit_code": exit_code}),
         );
         if process::stopping() || config.max_runs.is_some_and(|n| completed >= n) {
             break;
         }
         status(root, "sleeping", seq, completed)?;
+        emit(Level::Info, "daemon_sleeping", || {
+            json!({"interval_secs": config.interval,
+                "next_run_at_unix": time::OffsetDateTime::now_utc().unix_timestamp().checked_add(i64::try_from(config.interval).unwrap_or(i64::MAX))})
+        });
         process::sleep(config.interval);
     }
     status(root, "exited", lifecycle.last_seq, completed)?;
+    emit(
+        Level::Info,
+        "daemon_exited",
+        || json!({"runs_completed": completed, "failed_runs": failures, "stop_requested": process::stopping()}),
+    );
     if failures > 0 {
         Err(format!("daemon completed with {failures} failed run(s)"))
     } else {

src/daemon/process.rs

@@ -68,9 +68,15 @@ pub(super) fn execute(
             .map_err(|e| format!("spawn validator: {e}"))?,
     );
     let start = Instant::now();
+    crate::logging::emit(
+        crate::logging::Level::Info,
+        "daemon_child_started",
+        || serde_json::json!({"pid": child.0.id()}),
+    );
     let mut stop_at = None;
     let mut terminated_at = None;
     let mut reason = None;
+    let mut killed = false;
     loop {
         if let Some(status) = child
             .0
@@ -80,17 +86,35 @@ pub(super) fn execute(
             return Ok((status, reason));
         }
         if stopping() {
+            if stop_at.is_none() {
+                crate::logging::emit(
+                    crate::logging::Level::Info,
+                    "daemon_stop_requested",
+                    || serde_json::json!({"grace_secs": grace}),
+                );
+            }
             stop_at.get_or_insert_with(Instant::now);
         }
         let timed_out = timeout > 0 && start.elapsed().as_secs() >= timeout;
         let grace_expired = stop_at.is_some_and(|at: Instant| at.elapsed().as_secs() >= grace);
         if terminated_at.is_none() && (timed_out || grace_expired) {
             reason = Some(if timed_out { "timeout" } else { "interrupted" });
+            crate::logging::emit(
+                crate::logging::Level::Warn,
+                "daemon_child_terminate",
+                || serde_json::json!({"reason": reason, "signal": "SIGTERM", "elapsed_ms": start.elapsed().as_millis()}),
+            );
             child.signal_group(libc::SIGTERM);
             terminated_at = Some(Instant::now());
         }
-        if terminated_at.is_some_and(|at| at.elapsed() >= Duration::from_secs(2)) {
+        if !killed && terminated_at.is_some_and(|at| at.elapsed() >= Duration::from_secs(2)) {
+            crate::logging::emit(
+                crate::logging::Level::Warn,
+                "daemon_child_kill",
+                || serde_json::json!({"reason": reason, "signal": "SIGKILL"}),
+            );
             child.signal_group(libc::SIGKILL);
+            killed = true;
         }
         std::thread::sleep(Duration::from_millis(50));
     }

src/logging.rs

@@ -2,6 +2,17 @@
 use std::io::Write;
 use std::sync::Mutex;
 use std::sync::atomic::{AtomicU8, Ordering};
+pub(crate) mod artifact;
+pub(crate) mod observer;
+mod text;
+
+static CONTEXT: Mutex<Option<serde_json::Value>> = Mutex::new(None);
+
+pub(crate) fn set_context(run_id: &str, run_seq: Option<u64>) {
+    *CONTEXT.lock().unwrap_or_else(|error| error.into_inner()) = Some(serde_json::json!({
+        "run_id": run_id, "run_seq": run_seq,
+    }));
+}
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[repr(u8)]
@@ -47,7 +58,7 @@ pub fn enabled(level: Level) -> bool {
     level as u8 <= FILTER.load(Ordering::Relaxed)
 }
 
-fn redact_text(text: &str) -> String {
+pub(crate) fn redact_text(text: &str) -> String {
     let mut output = String::with_capacity(text.len());
     let lower = text.to_ascii_lowercase();
     let mut cursor = 0;
@@ -123,6 +134,7 @@ fn redact(value: &mut serde_json::Value) {
 }
 
 /// Payload creation is lazy, so disabled diagnostics do not allocate JSON.
+#[track_caller]
 pub fn emit(level: Level, event: &str, payload: impl FnOnce() -> serde_json::Value) {
     let _ = emit_to(
         FILTER.load(Ordering::Relaxed),
@@ -139,6 +151,7 @@ pub fn emit(level: Level, event: &str, payload: impl FnOnce() -> serde_json::Val
     );
 }
 
+#[track_caller]
 fn emit_to(
     filter: u8,
     json: bool,
@@ -154,22 +167,36 @@ fn emit_to(
         .format(&time::format_description::well_known::Rfc3339)
         .unwrap_or_default();
     let mut fields = payload();
+    if let Some(context) = CONTEXT
+        .lock()
+        .unwrap_or_else(|error| error.into_inner())
+        .as_ref()
+        && let Some(fields) = fields.as_object_mut()
+        && let Some(context) = context.as_object()
+    {
+        fields.extend(context.clone());
+    }
     redact(&mut fields);
+    let source = std::panic::Location::caller();
     let record = serde_json::json!({
         "timestamp": timestamp,
         "level": format!("{level:?}").to_lowercase(),
         "event": event,
         "fields": fields,
+        "source": {"file": source.file(), "line": source.line()},
     });
     let line = if json {
         format!("{record}\n")
     } else {
+        if event == "validation_stage_completed" && record["fields"]["phase"] == "initializing" {
+            return Ok(());
+        }
+        let message = text::render(&timestamp, level, event, &record["fields"]);
         format!(
-            "{} {} {} {}\n",
+            "{} [{}:{}]\n",
-            timestamp,
+            message.trim_end_matches('\n'),
-            record["level"].as_str().unwrap_or(""),
+            source.file(),
-            event,
+            source.line()
-            record["fields"]
         )
     };
     sink(line.as_bytes())
@@ -179,6 +206,76 @@ fn emit_to(
 mod tests {
     use super::*;
 
+    #[track_caller]
+    fn forwarded_record(json: bool) -> String {
+        let mut output = String::new();
+        emit_to(
+            3,
+            json,
+            Level::Info,
+            "validation_completed",
+            || serde_json::json!({"elapsed_ms":1000}),
+            |line| {
+                output = String::from_utf8(line.to_vec()).unwrap();
+                Ok(())
+            },
+        )
+        .unwrap();
+        output
+    }
+
+    #[test]
+    fn source_location_survives_forwarding_and_matches_both_formats() {
+        let expected_line = line!() + 1;
+        let raw = forwarded_record(true);
+        let record: serde_json::Value = serde_json::from_str(&raw).unwrap();
+        assert_eq!(record["source"]["file"], file!());
+        assert_eq!(record["source"]["line"], expected_line);
+        let expected_line = line!() + 1;
+        let text = forwarded_record(false);
+        assert!(text.ends_with(&format!("[{}:{expected_line}]\n", file!())));
+    }
+
+    #[test]
+    fn text_redacts_before_rendering_and_json_keeps_initializing() {
+        emit_to(
+            3,
+            false,
+            Level::Warn,
+            "http_fetch_failed",
+            || {
+                serde_json::json!({
+                    "error": "fetch https://alice:pw@example.test/a?private=yes\nfailed"
+                })
+            },
+            |line| {
+                let line = std::str::from_utf8(line).unwrap();
+                assert!(line.contains("WARN HTTP download failed"));
+                assert!(!line.contains("alice"));
+                assert!(!line.contains("private=yes"));
+                assert_eq!(line.lines().count(), 1);
+                Ok(())
+            },
+        )
+        .unwrap();
+        for json in [false, true] {
+            let written = std::cell::Cell::new(false);
+            emit_to(
+                3,
+                json,
+                Level::Info,
+                "validation_stage_completed",
+                || serde_json::json!({"phase": "initializing", "elapsed_ms": 0}),
+                |_| {
+                    written.set(true);
+                    Ok(())
+                },
+            )
+            .unwrap();
+            assert_eq!(written.get(), json);
+        }
+    }
+
     #[test]
     fn mixed_case_urls_and_opaque_credentials_are_redacted() {
         assert_eq!(
@@ -320,13 +417,6 @@ mod tests {
 }
 pub(crate) mod progress;
 
-macro_rules! info {
-    ($($args:tt)*) => {
-        crate::logging::emit(crate::logging::Level::Info, "runtime", || serde_json::json!({"message": format!($($args)*)}))
-    };
-}
-pub(crate) use info;
-
 macro_rules! warning {
     ($($args:tt)*) => {
         crate::logging::emit(crate::logging::Level::Warn, "runtime_warning", || serde_json::json!({"message": format!($($args)*)}))

src/logging/artifact.rs

@@ -0,0 +1,34 @@
+use super::{Level, emit};
+use serde_json::json;
+use std::path::Path;
+use std::time::Instant;
+
+pub(crate) fn write<T>(
+    kind: &str,
+    path: &Path,
+    operation: impl FnOnce() -> Result<T, String>,
+) -> Result<T, String> {
+    let started = Instant::now();
+    emit(
+        Level::Info,
+        "artifact_write_started",
+        || json!({"artifact_type": kind, "path": path}),
+    );
+    match operation() {
+        Ok(value) => {
+            emit(Level::Info, "artifact_write_completed", || {
+                json!({"artifact_type": kind,
+                "path": path, "bytes": std::fs::metadata(path).ok().map(|meta| meta.len()),
+                "elapsed_ms": started.elapsed().as_millis()})
+            });
+            Ok(value)
+        }
+        Err(error) => {
+            emit(Level::Error, "artifact_write_failed", || {
+                json!({"artifact_type": kind,
+                "path": path, "error": error, "elapsed_ms": started.elapsed().as_millis()})
+            });
+            Err(error)
+        }
+    }
+}

src/logging/observer.rs

@@ -0,0 +1,297 @@
+use super::{Level, emit};
+use serde_json::{Value, json};
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, mpsc};
+use std::time::{Duration, Instant};
+
+static ACTIVE: Mutex<Option<Arc<Mutex<Snapshot>>>> = Mutex::new(None);
+static ENABLED: AtomicBool = AtomicBool::new(false);
+
+struct Snapshot {
+    started: Instant,
+    phase_started: Instant,
+    updated: Instant,
+    advanced: Instant,
+    phase: &'static str,
+    completed: u64,
+    counters: Value,
+    repositories: HashMap<String, Instant>,
+    repositories_completed: u64,
+    repositories_failed: u64,
+    objects_written: u64,
+    downloaded_bytes: u64,
+    roa_tasks_completed: u64,
+    slow_operations: u64,
+    slowest_ms: u64,
+    slowest_event: String,
+}
+
+impl Snapshot {
+    fn record(&self, now: Instant) -> Value {
+        json!({
+            "phase": self.phase,
+            "elapsed_ms": now.duration_since(self.started).as_millis(),
+            "phase_elapsed_ms": now.duration_since(self.phase_started).as_millis(),
+            "statistics_age_ms": now.duration_since(self.updated).as_millis(),
+            "progress_stalled": now.duration_since(self.advanced) >= Duration::from_secs(30),
+            "counters": self.counters,
+            "repository_attempts_inflight": self.repositories.len(),
+            "repository_attempts_completed": self.repositories_completed,
+            "repository_attempts_failed": self.repositories_failed,
+            "repository_objects_written": self.objects_written,
+            "downloaded_snapshot_delta_bytes": self.downloaded_bytes,
+            "roa_tasks_completed": self.roa_tasks_completed,
+            "slow_operations": self.slow_operations,
+            "slowest_ms": self.slowest_ms,
+            "slowest_event": self.slowest_event,
+            "longest_repository_wait_ms": self.repositories.values().map(|started| now.duration_since(*started).as_millis()).max(),
+        })
+    }
+}
+
+pub(crate) struct Observer {
+    stop: mpsc::Sender<()>,
+    worker: Option<std::thread::JoinHandle<()>>,
+}
+
+impl Observer {
+    pub(crate) fn finish(mut self, success: bool) {
+        let _ = self.stop.send(());
+        if let Some(worker) = self.worker.take() {
+            let _ = worker.join();
+        }
+        if let Some(shared) = ACTIVE
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .as_ref()
+        {
+            let record = shared
+                .lock()
+                .unwrap_or_else(|error| error.into_inner())
+                .record(Instant::now());
+            emit(
+                if success { Level::Info } else { Level::Error },
+                if success {
+                    "validation_stage_completed"
+                } else {
+                    "validation_stage_failed"
+                },
+                || record,
+            );
+        }
+    }
+
+    pub(crate) fn start(seconds: u64) -> Result<Self, String> {
+        let now = Instant::now();
+        let snapshot = Arc::new(Mutex::new(Snapshot {
+            started: now,
+            phase_started: now,
+            updated: now,
+            advanced: now,
+            phase: "initializing",
+            completed: 0,
+            counters: json!({}),
+            repositories: HashMap::new(),
+            repositories_completed: 0,
+            repositories_failed: 0,
+            objects_written: 0,
+            downloaded_bytes: 0,
+            roa_tasks_completed: 0,
+            slow_operations: 0,
+            slowest_ms: 0,
+            slowest_event: String::new(),
+        }));
+        let (stop, receiver) = mpsc::channel();
+        let worker = if seconds > 0 && super::enabled(Level::Info) {
+            let shared = Arc::clone(&snapshot);
+            Some(
+                std::thread::Builder::new()
+                    .name("validation-progress".into())
+                    .spawn(move || {
+                        while receiver.recv_timeout(Duration::from_secs(seconds))
+                            == Err(mpsc::RecvTimeoutError::Timeout)
+                        {
+                            let record = shared
+                                .lock()
+                                .unwrap_or_else(|error| error.into_inner())
+                                .record(Instant::now());
+                            emit(Level::Info, "validation_progress", || record);
+                        }
+                    })
+                    .map_err(|error| format!("start progress observer: {error}"))?,
+            )
+        } else {
+            None
+        };
+        if super::enabled(Level::Info) {
+            *ACTIVE.lock().unwrap_or_else(|error| error.into_inner()) = Some(snapshot);
+            ENABLED.store(true, Ordering::Relaxed);
+        }
+        Ok(Self { stop, worker })
+    }
+}
+
+impl Drop for Observer {
+    fn drop(&mut self) {
+        ENABLED.store(false, Ordering::Relaxed);
+        let _ = self.stop.send(());
+        if let Some(worker) = self.worker.take() {
+            let _ = worker.join();
+        }
+        *ACTIVE.lock().unwrap_or_else(|error| error.into_inner()) = None;
+    }
+}
+
+pub(crate) fn phase(name: &'static str) {
+    if let Some(shared) = ACTIVE
+        .lock()
+        .unwrap_or_else(|error| error.into_inner())
+        .as_ref()
+    {
+        let mut snapshot = shared.lock().unwrap_or_else(|error| error.into_inner());
+        let now = Instant::now();
+        emit(Level::Info, "validation_stage_completed", || {
+            snapshot.record(now)
+        });
+        snapshot.phase = name;
+        snapshot.phase_started = now;
+        snapshot.updated = now;
+        snapshot.advanced = now;
+        snapshot.completed = 0;
+        snapshot.counters = json!({});
+        emit(Level::Info, "validation_stage_started", || {
+            snapshot.record(now)
+        });
+    }
+}
+
+pub(crate) fn update(completed: u64, counters: impl FnOnce() -> Value) {
+    if !ENABLED.load(Ordering::Relaxed) {
+        return;
+    }
+    if let Some(shared) = ACTIVE
+        .lock()
+        .unwrap_or_else(|error| error.into_inner())
+        .as_ref()
+    {
+        let mut snapshot = shared.lock().unwrap_or_else(|error| error.into_inner());
+        let now = Instant::now();
+        if now.duration_since(snapshot.updated) < Duration::from_millis(250) {
+            return;
+        }
+        if completed != snapshot.completed {
+            snapshot.advanced = now;
+        }
+        snapshot.completed = completed;
+        snapshot.updated = now;
+        snapshot.counters = counters();
+    }
+}
+
+pub(crate) fn observes(kind: &str) -> bool {
+    ENABLED.load(Ordering::Relaxed)
+        && matches!(
+            kind,
+            "rrdp_sync_attempt"
+                | "rrdp_sync_success"
+                | "rrdp_sync_failed"
+                | "phase2_object_results_drain"
+                | "rrdp_download_completed"
+                | "http_fetch_slow"
+                | "rsync_sync_slow"
+                | "phase2_control_loop_slow"
+                | "phase2_stage_fresh_slow"
+                | "phase2_ready_publication_point_control_slow"
+        )
+}
+
+pub(crate) fn event(kind: &str, fields: &Value) {
+    if let Some(shared) = ACTIVE
+        .lock()
+        .unwrap_or_else(|error| error.into_inner())
+        .as_ref()
+    {
+        let mut snapshot = shared.lock().unwrap_or_else(|error| error.into_inner());
+        let now = Instant::now();
+        match kind {
+            "rrdp_download_completed" => {
+                snapshot.downloaded_bytes += fields["bytes"].as_u64().unwrap_or(0);
+                snapshot.advanced = now;
+            }
+            "rrdp_sync_attempt" => {
+                if let Some(uri) = fields["notify_uri"].as_str() {
+                    snapshot.repositories.insert(uri.into(), now);
+                }
+            }
+            "rrdp_sync_success" | "rrdp_sync_failed" => {
+                if let Some(uri) = fields["notify_uri"].as_str() {
+                    snapshot.repositories.remove(uri);
+                }
+                snapshot.repositories_completed += 1;
+                snapshot.repositories_failed += u64::from(kind == "rrdp_sync_failed");
+                snapshot.objects_written += fields["objects_written"].as_u64().unwrap_or(0);
+                snapshot.advanced = now;
+            }
+            "phase2_object_results_drain" => {
+                let completed = fields["results_drained"].as_u64().unwrap_or(0);
+                snapshot.roa_tasks_completed += completed;
+                if completed > 0 {
+                    snapshot.advanced = now;
+                }
+            }
+            _ => {
+                snapshot.slow_operations += 1;
+                let elapsed = [
+                    "duration_ms",
+                    "total_duration_ms",
+                    "stage_fresh_ms",
+                    "elapsed_ms",
+                    "control_ms",
+                ]
+                .iter()
+                .filter_map(|key| fields[key].as_u64())
+                .max()
+                .unwrap_or(0);
+                if elapsed > snapshot.slowest_ms {
+                    snapshot.slowest_ms = elapsed;
+                    snapshot.slowest_event = kind.into();
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn stalled_and_statistics_age_are_explicit() {
+        let now = Instant::now();
+        let snapshot = Snapshot {
+            started: now,
+            phase_started: now,
+            updated: now,
+            advanced: now,
+            phase: "online_validation",
+            completed: 0,
+            counters: json!({"publication_points_completed": 0}),
+            repositories: HashMap::new(),
+            repositories_completed: 0,
+            repositories_failed: 0,
+            objects_written: 0,
+            downloaded_bytes: 0,
+            roa_tasks_completed: 0,
+            slow_operations: 0,
+            slowest_ms: 0,
+            slowest_event: String::new(),
+        };
+        assert_eq!(
+            snapshot.record(now + Duration::from_secs(29))["progress_stalled"],
+            false
+        );
+        let record = snapshot.record(now + Duration::from_secs(30));
+        assert_eq!(record["progress_stalled"], true);
+        assert_eq!(record["statistics_age_ms"], 30000);
+    }
+}

src/logging/progress.rs

@@ -1,41 +1,102 @@
 // Event sites construct detailed payloads lazily.
 macro_rules! emit {
     ($kind:expr, $payload:expr $(,)?) => {
-        crate::logging::emit(crate::logging::progress::event_level($kind), $kind, || {
+        crate::logging::progress::emit_event($kind, || $payload)
-            $payload
-        })
     };
 }
 pub(crate) use emit;
 
-pub(crate) fn event_level(kind: &str) -> crate::logging::Level {
+#[track_caller]
-    if matches!(kind, "http_fetch_failed" | "rrdp_sync_failed") {
+pub(crate) fn emit_event(kind: &str, payload: impl FnOnce() -> serde_json::Value) {
-        crate::logging::Level::Warn
+    let level = event_level(kind);
-    } else if kind == "validation_failed" || kind.ends_with("_failed") || kind.starts_with("error_")
+    if super::observer::observes(kind) {
-    {
+        let fields = payload();
-        crate::logging::Level::Error
+        super::observer::event(kind, &fields);
-    } else if kind.ends_with("_warning")
+        super::emit(level, kind, || fields);
-        || kind.ends_with("_warnings")
-        || kind.starts_with("warning_")
-    {
-        crate::logging::Level::Warn
-    } else if kind.starts_with("publication_point")
-        || kind.starts_with("phase2_")
-        || kind.starts_with("object_")
-    {
-        crate::logging::Level::Trace
-    } else if kind.starts_with("validation_")
-        || kind.starts_with("run_")
-        || kind.starts_with("output_")
-        || kind.starts_with("ccr_")
-        || kind.ends_with("_summary")
-    {
-        crate::logging::Level::Info
     } else {
-        crate::logging::Level::Debug
+        super::emit(level, kind, payload);
     }
 }
 
+pub(crate) fn event_level(kind: &str) -> crate::logging::Level {
+    registered_event_level(kind).unwrap_or(crate::logging::Level::Debug)
+}
+
+fn registered_event_level(kind: &str) -> Option<crate::logging::Level> {
+    Some(
+        if matches!(
+            kind,
+            "http_fetch_failed"
+                | "rrdp_sync_failed"
+                | "rrdp_cross_origin_rejected"
+                | "repo_terminal_failure"
+                | "rrdp_failed_fallback_rsync"
+        ) {
+            crate::logging::Level::Warn
+        } else if matches!(kind, "validation_failed" | "object_failed") {
+            crate::logging::Level::Error
+        } else if matches!(kind, "validation_warning" | "validation_warnings") {
+            crate::logging::Level::Warn
+        } else if matches!(
+            kind,
+            "publication_point_start"
+                | "publication_point_repo_sync_done"
+                | "publication_point_finish"
+                | "publication_point_object"
+                | "phase2_repo_events_drain"
+                | "phase2_ready_queue_stage_fresh_breakdown"
+                | "phase2_stage_pool_stats"
+                | "phase2_publication_point_reduced"
+                | "phase2_build_tree_output"
+                | "phase2_roa_dispatch_batch"
+                | "phase2_object_results_drain"
+                | "phase2_finalize_task_submit"
+                | "phase2_finalize_results_drain"
+                | "phase2_ready_queue_batch"
+                | "phase2_ready_queue_control_breakdown"
+                | "phase2_ready_publication_point_control_snapshot_breakdown"
+                | "phase2_finalize_worker_breakdown"
+        ) {
+            crate::logging::Level::Trace
+        } else if matches!(
+            kind,
+            "validation_started" | "validation_completed" | "validation_summary"
+        ) {
+            crate::logging::Level::Info
+        } else if matches!(
+            kind,
+            "repository_sync_selected"
+                | "repo_sync_rrdp_ok"
+                | "rrdp_fallback_rsync"
+                | "repo_sync_rsync_direct"
+                | "rrdp_sync_attempt"
+                | "rrdp_sync_success"
+                | "rsync_sync_start"
+                | "rsync_sync_fetch_done"
+                | "rsync_sync_done"
+                | "rsync_sync_slow"
+                | "http_fetch_slow"
+                | "rrdp_redirect_checked"
+                | "phase2_control_loop_slow"
+                | "phase2_stage_fresh_slow"
+                | "phase2_ready_publication_point_control_slow"
+                | "phase1_repo_task_enqueued"
+                | "phase1_repo_task_waiting"
+                | "phase1_repo_task_reused"
+                | "phase1_repo_task_dispatched"
+                | "phase1_repo_task_result"
+                | "phase1_repo_worker_execute_start"
+                | "phase1_repo_worker_execute_finish"
+                | "rrdp_sync_selected"
+                | "rrdp_download_completed"
+        ) {
+            crate::logging::Level::Debug
+        } else {
+            return None;
+        },
+    )
+}
+
 pub fn slow_threshold_secs() -> f64 {
     30.0
 }
@@ -57,6 +118,37 @@ mod tests {
     use super::event_level;
     use crate::logging::Level;
 
+    #[test]
+    fn all_literal_progress_sites_are_registered() {
+        fn visit(path: &std::path::Path) {
+            for entry in std::fs::read_dir(path).unwrap() {
+                let path = entry.unwrap().path();
+                if path.is_dir() {
+                    visit(&path);
+                } else if path.extension().is_some_and(|extension| extension == "rs") {
+                    let source = std::fs::read_to_string(&path).unwrap();
+                    for site in source.split(concat!("progress::", "emit!(")).skip(1) {
+                        if let Some(literal) = site.trim_start().strip_prefix('"') {
+                            let event = literal.split('"').next().unwrap();
+                            assert!(
+                                super::registered_event_level(event).is_some(),
+                                "unregistered event {event} in {}",
+                                path.display()
+                            );
+                        }
+                    }
+                }
+            }
+        }
+        visit(&std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"));
+        assert_eq!(
+            super::registered_event_level("phase2_finalize_worker_breakdown"),
+            Some(Level::Trace)
+        );
+        assert_eq!(event_level("rrdp_cross_origin_rejected"), Level::Warn);
+        assert_eq!(event_level("phase2_control_loop_slow"), Level::Debug);
+    }
+
     #[test]
     fn event_levels_keep_default_output_quiet() {
         assert_eq!(event_level("validation_started"), Level::Info);

src/logging/text.rs

@@ -0,0 +1,506 @@
+use super::Level;
+use serde_json::Value;
+
+fn string<'a>(fields: &'a Value, key: &str, fallback: &'a str) -> &'a str {
+    fields[key]
+        .as_str()
+        .filter(|value| !value.is_empty())
+        .unwrap_or(fallback)
+}
+
+fn number(fields: &Value, key: &str) -> String {
+    fields[key]
+        .as_u64()
+        .map(|value| value.to_string())
+        .unwrap_or_else(|| "an unknown number of".into())
+}
+
+fn duration(fields: &Value, key: &str) -> String {
+    fields[key]
+        .as_u64()
+        .map(|value| format!("{:.1} seconds", value as f64 / 1000.0))
+        .unwrap_or_else(|| "an unknown duration".into())
+}
+
+fn reason(fields: &Value) -> String {
+    let raw = string(
+        fields,
+        "error",
+        string(
+            fields,
+            "message",
+            string(fields, "rrdp_error", "No further details were reported."),
+        ),
+    );
+    let shortened: String = raw.chars().take(240).collect();
+    if raw.chars().count() > 240 {
+        format!("{shortened}… (use JSON logging for the full diagnostic)")
+    } else {
+        shortened
+    }
+}
+
+fn repository(fields: &Value) -> &str {
+    string(
+        fields,
+        "notify_uri",
+        string(
+            fields,
+            "manifest_rsync_uri",
+            string(fields, "uri", "the repository"),
+        ),
+    )
+}
+
+fn phase(fields: &Value) -> (&'static str, &'static str) {
+    match fields["phase"].as_str() {
+        Some("state_store_open") => (
+            "Opening the saved repository state",
+            "Opened the saved repository state",
+        ),
+        Some("runtime_setup") => ("Preparing the validator", "Prepared the validator"),
+        Some("trust_anchor_discovery") => ("Loading trust anchors", "Loaded trust anchors"),
+        Some("online_validation") => (
+            "Synchronizing repositories and validating objects",
+            "Finished synchronizing repositories and validating objects",
+        ),
+        Some("result_output") => ("Writing validation results", "Wrote validation results"),
+        Some("vrp_csv_conversion") => ("Exporting VRPs to CSV", "Exported VRPs to CSV"),
+        _ => ("Initializing validation", "Finished initialization"),
+    }
+}
+
+fn progress(fields: &Value) -> String {
+    let mut sentence = format!(
+        "{}; {} elapsed.",
+        phase(fields).0,
+        duration(fields, "elapsed_ms")
+    );
+    if fields["phase"] == "online_validation" {
+        if let Some(completed) = fields["counters"]["publication_points_completed"].as_u64() {
+            let stale = fields["statistics_age_ms"]
+                .as_u64()
+                .is_some_and(|age| age > 2000);
+            sentence.push_str(&if stale {
+                format!(" The last update reported {completed} publication points processed.")
+            } else {
+                format!(" Processed {completed} publication points so far.")
+            });
+        }
+        if let Some(inflight) = fields["repository_attempts_inflight"]
+            .as_u64()
+            .filter(|count| *count > 0)
+        {
+            sentence.push_str(&if inflight == 1 {
+                " 1 repository sync attempt is still in progress.".into()
+            } else {
+                format!(" {inflight} repository sync attempts are still in progress.")
+            });
+        }
+    }
+    if fields["progress_stalled"] == true {
+        sentence.push_str(" No completed work has been observed for at least 30 seconds; this does not necessarily mean the validator is stuck.");
+    }
+    sentence
+}
+
+fn category(fields: &Value) -> &'static str {
+    match fields["category"].as_str() {
+        Some("transport") => "network access",
+        Some("rrdp_protocol") => "RRDP protocol checks",
+        Some("security_policy") => "security policy checks",
+        Some("certificate") => "certificate validation",
+        Some("manifest_crl") => "manifest and CRL checks",
+        Some("object_validation") => "signed object validation",
+        Some("ber_compatible_cms_encoding") => "accepted BER-compatible CMS encoding",
+        _ => "issues without a recorded category",
+    }
+}
+
+fn message(event: &str, fields: &Value) -> String {
+    match event {
+        "validation_started" => format!(
+            "Starting validation with {} synchronization across {} trust anchors.",
+            string(fields, "sync_mode", "the configured"),
+            number(fields, "tal_count")
+        ),
+        "validation_completed" => format!(
+            "Validation completed in {}.",
+            duration(fields, "elapsed_ms")
+        ),
+        "validation_failed" => format!(
+            "Validation failed after {}: {}",
+            duration(fields, "elapsed_ms"),
+            reason(fields)
+        ),
+        "command_failed" => format!("Command failed: {}", reason(fields)),
+        "validation_summary" => format!(
+            "Validation produced {} VRPs and {} ASPA records from {} publication points.",
+            number(fields, "vrps"),
+            number(fields, "aspas"),
+            number(fields, "publication_points")
+        ),
+        "validation_progress" => progress(fields),
+        "validation_stage_started" => format!("{}.", phase(fields).0),
+        "validation_stage_completed" => format!(
+            "{} in {}.",
+            phase(fields).1,
+            duration(fields, "phase_elapsed_ms")
+        ),
+        "validation_stage_failed" => format!(
+            "Validation stopped while {} after {}.",
+            phase(fields).0.to_ascii_lowercase(),
+            duration(fields, "phase_elapsed_ms")
+        ),
+        "trust_anchor_started" => format!(
+            "Loading trust anchor {}.",
+            string(fields, "tal_id", "with an unknown ID")
+        ),
+        "trust_anchor_completed" => format!(
+            "Trust anchor {} is ready.",
+            string(fields, "tal_id", "with an unknown ID")
+        ),
+        "trust_anchor_failed" => format!(
+            "Could not load trust anchor {}: {}",
+            string(fields, "tal_id", "with an unknown ID"),
+            reason(fields)
+        ),
+        "artifact_write_started" => {
+            format!("Writing {}.", string(fields, "path", "the output file"))
+        }
+        "artifact_write_completed" => format!(
+            "Saved {} in {}.",
+            string(fields, "path", "the output file"),
+            duration(fields, "elapsed_ms")
+        ),
+        "artifact_write_failed" => format!(
+            "Could not write {}: {}",
+            string(fields, "path", "the output file"),
+            reason(fields)
+        ),
+        "validation_warnings" => format!(
+            "Validation reported {} warnings. Review the affected repositories and objects before interpreting the results.",
+            number(fields, "count")
+        ),
+        "validation_warning_category" => format!(
+            "Recorded {} warnings about {}.",
+            number(fields, "count"),
+            category(fields)
+        ),
+        "validation_warning_detail" | "validation_warning" | "runtime_warning" => reason(fields),
+        "object_failed" => format!("Object validation failed: {}", reason(fields)),
+        "rrdp_sync_attempt" => format!("Fetching repository updates from {}.", repository(fields)),
+        "rrdp_sync_success" => format!(
+            "Updated {} in {}.",
+            repository(fields),
+            duration(fields, "duration_ms")
+        ),
+        "rrdp_sync_failed" => format!(
+            "Could not update {} using RRDP: {}",
+            repository(fields),
+            reason(fields)
+        ),
+        "repo_terminal_failure" => format!(
+            "Could not process {}: {}",
+            repository(fields),
+            reason(fields)
+        ),
+        "http_fetch_failed" => format!(
+            "HTTP download failed for {}: {}",
+            repository(fields),
+            reason(fields)
+        ),
+        "rrdp_delta_fallback" => format!(
+            "Could not apply incremental updates from {}; trying a snapshot instead. {}",
+            repository(fields),
+            reason(fields)
+        ),
+        "rrdp_cross_origin_rejected" => format!(
+            "Rejected the RRDP {} because its origin ({}) differs from the notification origin ({}).",
+            string(fields, "resource", "resource"),
+            string(fields, "resource_origin", "unknown"),
+            string(fields, "notification_origin", "unknown")
+        ),
+        "rrdp_failed_fallback_rsync" => format!(
+            "RRDP failed for {}; trying rsync instead. {}",
+            repository(fields),
+            reason(fields)
+        ),
+        "rrdp_sync_selected" => format!(
+            "Using {} synchronization for {}.",
+            string(fields, "mode", "the selected"),
+            repository(fields)
+        ),
+        "rrdp_download_completed" => format!(
+            "Finished downloading repository data from {}.",
+            repository(fields)
+        ),
+        "daemon_started" => format!(
+            "Daemon started. Runs will be stored in {}; waiting {} seconds between runs.",
+            string(fields, "state_root", "the configured directory"),
+            number(fields, "interval_secs")
+        ),
+        "daemon_state_recovered" => format!(
+            "Resuming saved daemon state after run {}.",
+            number(fields, "last_run_seq")
+        ),
+        "daemon_run_started" => format!(
+            "Starting the next validation run. Follow its progress in {}.",
+            string(fields, "validator_log", "validator.log")
+        ),
+        "daemon_child_started" => "The validator process has started.".into(),
+        "daemon_run_completed" if fields["success"] == true => format!(
+            "Run completed successfully in {}.",
+            duration(fields, "wall_ms")
+        ),
+        "daemon_run_completed" => format!(
+            "Run failed after {}: {}",
+            duration(fields, "wall_ms"),
+            reason(fields)
+        ),
+        "daemon_sleeping" => format!(
+            "The next run will start in {} seconds.",
+            number(fields, "interval_secs")
+        ),
+        "daemon_retention_applied" => format!(
+            "Removed {} old run directories; keeping the latest {} runs.",
+            fields["removed_runs"]
+                .as_array()
+                .map(|runs| runs.len().to_string())
+                .unwrap_or_else(|| "an unknown number of".into()),
+            number(fields, "retain_runs")
+        ),
+        "daemon_exited" => format!(
+            "Daemon stopped after {} runs, including {} failed runs.",
+            number(fields, "runs_completed"),
+            number(fields, "failed_runs")
+        ),
+        "daemon_stop_requested" => "Shutdown requested. Waiting for the validator to stop.".into(),
+        "daemon_child_terminate" => "Asking the validator process to terminate.".into(),
+        "daemon_child_kill" => {
+            "The validator did not stop within the grace period; forcing it to exit.".into()
+        }
+        _ => {
+            let description = if ["error", "message", "rrdp_error"]
+                .iter()
+                .any(|key| fields[*key].as_str().is_some())
+            {
+                reason(fields)
+            } else {
+                "An internal diagnostic was recorded; use JSON logging for details.".into()
+            };
+            format!("{description} [diagnostic: {event}]")
+        }
+    }
+}
+
+pub(super) fn render(timestamp: &str, level: Level, event: &str, fields: &Value) -> String {
+    let context = fields["run_seq"]
+        .as_u64()
+        .map(|sequence| format!(" [run {sequence}]"))
+        .unwrap_or_default();
+    let sentence = message(event, fields);
+    let escaped: String = sentence
+        .chars()
+        .flat_map(|character| {
+            if character.is_control() {
+                character.escape_default().collect::<Vec<_>>()
+            } else {
+                vec![character]
+            }
+        })
+        .collect();
+    format!(
+        "{timestamp} {}{context} {escaped}\n",
+        format!("{level:?}").to_uppercase()
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn starting_a_run_is_a_sentence_not_a_field_dump() {
+        let line = render(
+            "now",
+            Level::Info,
+            "validation_started",
+            &json!({
+                "run_seq":230,"sync_mode":"delta","tal_count":5,"run_id":"hidden-uuid",
+                "phase2_workers":8,"unexpected":{"secret_internal_detail":55}
+            }),
+        );
+        assert_eq!(
+            line,
+            "now INFO [run 230] Starting validation with delta synchronization across 5 trust anchors.\n"
+        );
+    }
+
+    #[test]
+    fn progress_selects_only_useful_facts() {
+        let fields = json!({"phase":"online_validation","elapsed_ms":10005,
+            "counters":{"publication_points_completed":32000,"ready_queue":999},
+            "repository_attempts_inflight":3,"repository_objects_written":456,
+            "statistics_age_ms":200,"slowest_event":"hidden","run_id":"hidden"});
+        assert_eq!(
+            message("validation_progress", &fields),
+            "Synchronizing repositories and validating objects; 10.0 seconds elapsed. Processed 32000 publication points so far. 3 repository sync attempts are still in progress."
+        );
+    }
+
+    #[test]
+    fn stale_and_stalled_progress_does_not_claim_deadlock_or_completion() {
+        let fields = json!({"phase":"online_validation","elapsed_ms":40000,
+            "counters":{"publication_points_completed":0},"statistics_age_ms":5000,"progress_stalled":true});
+        let sentence = progress(&fields);
+        assert!(sentence.contains("last update reported 0 publication points"));
+        assert!(sentence.contains("does not necessarily mean"));
+        assert!(!sentence.contains("sync attempts"));
+        assert!(
+            !progress(&json!({"phase":"state_store_open","elapsed_ms":10000}))
+                .contains("publication points")
+        );
+    }
+
+    #[test]
+    fn errors_are_bounded_and_cannot_inject_lines() {
+        let line = render(
+            "now",
+            Level::Warn,
+            "repo_terminal_failure",
+            &json!({"notify_uri":"https://example.test/notification.xml","error":format!("bad\n\r\u{1b}{}", "证书".repeat(300))}),
+        );
+        assert_eq!(line.lines().count(), 1);
+        assert!(line.contains("bad\\n\\r\\u{1b}"));
+        assert!(line.contains("use JSON logging"));
+        assert!(!line.contains("cached"));
+    }
+
+    #[test]
+    fn stages_outputs_and_results_have_independent_wording() {
+        assert_eq!(
+            message(
+                "validation_stage_completed",
+                &json!({"phase":"state_store_open","phase_elapsed_ms":339,"repository_objects_written":0})
+            ),
+            "Opened the saved repository state in 0.3 seconds."
+        );
+        assert_eq!(
+            message(
+                "artifact_write_completed",
+                &json!({"path":"/data/vrps.csv","elapsed_ms":133,"bytes":999})
+            ),
+            "Saved /data/vrps.csv in 0.1 seconds."
+        );
+        assert_eq!(
+            message(
+                "validation_summary",
+                &json!({"vrps":1012510,"aspas":3007,"publication_points":54463})
+            ),
+            "Validation produced 1012510 VRPs and 3007 ASPA records from 54463 publication points."
+        );
+        assert_eq!(
+            message(
+                "daemon_sleeping",
+                &json!({"interval_secs":60,"next_run_at_unix":123456})
+            ),
+            "The next run will start in 60 seconds."
+        );
+    }
+
+    #[test]
+    fn unknown_diagnostics_keep_the_message_not_arbitrary_fields() {
+        assert_eq!(
+            message(
+                "extension_notice",
+                &json!({"message":"A custom check failed.","arbitrary":99})
+            ),
+            "A custom check failed. [diagnostic: extension_notice]"
+        );
+        assert!(message("validation_completed", &json!({})).contains("unknown duration"));
+    }
+
+    #[test]
+    fn common_events_never_fall_back_to_technical_identifiers() {
+        let fields = json!({"error":"Connection refused.","notify_uri":"https://example.test/n.xml",
+            "duration_ms":1000,"elapsed_ms":1000,"phase_elapsed_ms":1000,"wall_ms":1000,
+            "tal_id":"ripe","path":"/data/result.ccr","count":2,"last_run_seq":229,
+            "validator_log":"/data/validator.log","removed_runs":["run_1"],"retain_runs":10,
+            "runs_completed":3,"failed_runs":0,"success":true,"state_root":"/data","interval_secs":60});
+        for event in [
+            "validation_failed",
+            "command_failed",
+            "validation_stage_started",
+            "validation_stage_failed",
+            "trust_anchor_started",
+            "trust_anchor_completed",
+            "trust_anchor_failed",
+            "artifact_write_started",
+            "artifact_write_failed",
+            "validation_warnings",
+            "validation_warning_detail",
+            "runtime_warning",
+            "rrdp_sync_attempt",
+            "rrdp_sync_success",
+            "rrdp_sync_failed",
+            "repo_terminal_failure",
+            "http_fetch_failed",
+            "rrdp_delta_fallback",
+            "rrdp_cross_origin_rejected",
+            "rrdp_sync_selected",
+            "rrdp_download_completed",
+            "daemon_started",
+            "daemon_state_recovered",
+            "daemon_run_started",
+            "daemon_child_started",
+            "daemon_run_completed",
+            "daemon_retention_applied",
+            "daemon_exited",
+            "daemon_stop_requested",
+            "daemon_child_terminate",
+            "daemon_child_kill",
+        ] {
+            let sentence = message(event, &fields);
+            assert!(!sentence.contains("diagnostic:"), "{event}: {sentence}");
+            assert!(!sentence.contains("run_id"));
+        }
+        assert!(
+            message(
+                "daemon_run_completed",
+                &json!({"success":false,"wall_ms":1000,"error":"Stopped."})
+            )
+            .starts_with("Run failed")
+        );
+        for name in [
+            "transport",
+            "rrdp_protocol",
+            "security_policy",
+            "certificate",
+            "manifest_crl",
+            "object_validation",
+            "ber_compatible_cms_encoding",
+            "unclassified",
+        ] {
+            assert!(
+                message(
+                    "validation_warning_category",
+                    &json!({"category":name,"count":2})
+                )
+                .starts_with("Recorded 2 warnings about ")
+            );
+        }
+        for name in [
+            "state_store_open",
+            "runtime_setup",
+            "trust_anchor_discovery",
+            "online_validation",
+            "result_output",
+            "vrp_csv_conversion",
+            "initializing",
+        ] {
+            assert!(!message("validation_stage_started", &json!({"phase":name})).contains('_'));
+        }
+    }
+}

src/output/report.rs

@@ -11,6 +11,12 @@ pub enum WarningCategory {
     #[default]
     Unclassified,
     BerCompatibleCmsEncoding,
+    Transport,
+    RrdpProtocol,
+    SecurityPolicy,
+    Certificate,
+    ManifestCrl,
+    ObjectValidation,
 }
 
 impl WarningCategory {
@@ -18,6 +24,12 @@ impl WarningCategory {
         match self {
             Self::Unclassified => "unclassified",
             Self::BerCompatibleCmsEncoding => "ber_compatible_cms_encoding",
+            Self::Transport => "transport",
+            Self::RrdpProtocol => "rrdp_protocol",
+            Self::SecurityPolicy => "security_policy",
+            Self::Certificate => "certificate",
+            Self::ManifestCrl => "manifest_crl",
+            Self::ObjectValidation => "object_validation",
         }
     }
 }

src/repository/sync/repo.rs

@@ -123,6 +123,20 @@ pub fn sync_publication_point(
                     }
                     let warnings = vec![
                         Warning::new(format!("RRDP failed; falling back to rsync: {err}"))
+                            .with_category(match &err {
+                                RrdpSyncError::Fetch(_) => {
+                                    crate::output::report::WarningCategory::Transport
+                                }
+                                RrdpSyncError::Rrdp(error) if error.is_cross_origin_violation() => {
+                                    crate::output::report::WarningCategory::SecurityPolicy
+                                }
+                                RrdpSyncError::Rrdp(_) => {
+                                    crate::output::report::WarningCategory::RrdpProtocol
+                                }
+                                RrdpSyncError::Storage(_) => {
+                                    crate::output::report::WarningCategory::Unclassified
+                                }
+                            })
                             .with_rfc_refs(&rfc_refs)
                             .with_context(notification_uri),
                     ];
@@ -281,6 +295,7 @@ fn try_rrdp_sync_with_retry(
     download_log: Option<&DownloadLogHandle>,
 ) -> Result<usize, RrdpSyncError> {
     let attempt = 1usize;
+    let started = std::time::Instant::now();
     crate::logging::progress::emit!(
         "rrdp_sync_attempt",
         serde_json::json!({
@@ -307,6 +322,7 @@ fn try_rrdp_sync_with_retry(
                     "notify_uri": notification_uri,
                     "attempt": attempt,
                     "objects_written": written,
+                    "duration_ms": started.elapsed().as_millis(),
                 }),
             );
             Ok(written)
@@ -322,6 +338,8 @@ fn try_rrdp_sync_with_retry(
                     "notify_uri": notification_uri,
                     "attempt": attempt,
                     "retryable": retryable,
+                    "will_retry": false,
+                    "duration_ms": started.elapsed().as_millis(),
                     "error": err.to_string(),
                 }),
             );

src/repository/sync/rrdp/notification_sync.rs

@@ -105,9 +105,20 @@ fn sync_from_notification_inner(
     let same_session_state = state
         .as_ref()
         .filter(|s| s.session_id == notif.session_id.to_string());
+    let log_selection = |mode: &str, reason: &str| {
+        crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || serde_json::json!({
+            "notify_uri": notification_uri, "mode": mode, "reason": reason,
+            "session_id": notif.session_id.to_string(), "serial": notif.serial,
+            "previous_serial": state.as_ref().map(|previous| previous.serial),
+            "previous_session_id": state.as_ref().map(|previous| &previous.session_id),
+        }));
+    };
+    let mut snapshot_reason = if state.is_none() { "no_local_state" }
+        else if same_session_state.is_none() { "session_changed" } else { "delta_chain_unavailable" };
 
     if let Some(s) = same_session_state {
         if s.serial == notif.serial {
+            log_selection("unchanged", "same_session_and_serial");
             let _hydrate_step = timing
                 .as_ref()
                 .map(|t| t.span_rrdp_repo_step(notification_uri, "hydrate_current_index"));
@@ -147,6 +158,7 @@ fn sync_from_notification_inner(
 
             // `parse_notification` guarantees contiguity and max==notif.serial when deltas exist.
             if max_serial == notif.serial && want_first >= min_serial {
+                log_selection("delta", "contiguous_delta_chain");
                 // Fetch all required delta files first so a network failure doesn't leave us with
                 // partially applied deltas and no snapshot fallback.
                 let _fetch_d_step = timing
@@ -179,6 +191,9 @@ fn sync_from_notification_inner(
                         &notification_origin,
                     ) {
                         Ok(bytes) => {
+                            crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({
+                                "resource": "delta", "bytes": bytes.len(), "notify_uri": notification_uri,
+                            }));
                             if let Some(t) = timing.as_ref() {
                                 t.record_count("rrdp_delta_fetch_ok_total", 1);
                                 t.record_count("rrdp_delta_bytes_total", bytes.len() as u64);
@@ -190,6 +205,7 @@ fn sync_from_notification_inner(
                             fetched.push((serial, dref.hash_sha256, bytes))
                         }
                         Err(RrdpFetchError::Fetch(e)) => {
+                            snapshot_reason = "delta_fetch_failed";
                             if let Some(t) = timing.as_ref() {
                                 t.record_count("rrdp_delta_fetch_fail_total", 1);
                             }
@@ -225,7 +241,9 @@ fn sync_from_notification_inner(
                             *serial,
                         ) {
                             Ok(n) => applied_total += n,
-                            Err(_) => {
+                            Err(error) => {
+                                snapshot_reason = "delta_apply_failed";
+                                crate::logging::emit(crate::logging::Level::Warn, "rrdp_delta_fallback", || serde_json::json!({"notify_uri": notification_uri, "error": error.to_string(), "serial": serial}));
                                 ok = false;
                                 break;
                             }
@@ -286,6 +304,7 @@ fn sync_from_notification_inner(
     }
 
     // Snapshot fallback (RFC 8182 §3.4.3).
+    log_selection("snapshot", snapshot_reason);
     let _fetch_step = timing
         .as_ref()
         .map(|t| t.span_rrdp_repo_step(notification_uri, "fetch_snapshot"));
@@ -301,6 +320,9 @@ fn sync_from_notification_inner(
         &notification_origin,
     ) {
         Ok(v) => {
+            crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({
+                "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri,
+            }));
             if let Some(t) = timing.as_ref() {
                 t.record_count("rrdp_snapshot_fetch_ok_total", 1);
                 t.record_count("rrdp_snapshot_bytes_total", v.1);

src/repository/sync/rrdp/snapshot_sync.rs

@@ -80,6 +80,10 @@ fn sync_from_notification_snapshot_inner(
     };
     drop(_parse_step);
     drop(_parse_total);
+    crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || serde_json::json!({
+        "notify_uri": notification_uri, "mode": "snapshot", "reason": "snapshot_requested",
+        "session_id": notif.session_id.to_string(), "serial": notif.serial,
+    }));
 
     let _fetch_step = timing
         .as_ref()
@@ -96,6 +100,9 @@ fn sync_from_notification_snapshot_inner(
         &notification_origin,
     ) {
         Ok(v) => {
+            crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({
+                "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri,
+            }));
             if let Some(t) = timing.as_ref() {
                 t.record_count("rrdp_snapshot_fetch_ok_total", 1);
                 t.record_count("rrdp_snapshot_bytes_total", v.1);

src/runtime/output.rs

@@ -320,14 +320,13 @@ pub(super) fn run_compare_view_task(
     if let (Some(vrps_path), Some(vaps_path)) = (vrps_csv_out_path, vaps_csv_out_path) {
         let started = std::time::Instant::now();
         build_ms = Some(0);
-        write_direct_vrp_csv(vrps_path, shared.vrps.as_ref(), trust_anchor)?;
+        crate::logging::artifact::write("vrps_source", vrps_path, || {
-        write_direct_vap_csv(vaps_path, shared.aspas.as_ref(), trust_anchor)?;
+            write_direct_vrp_csv(vrps_path, shared.vrps.as_ref(), trust_anchor)
+        })?;
+        crate::logging::artifact::write("vaps", vaps_path, || {
+            write_direct_vap_csv(vaps_path, shared.aspas.as_ref(), trust_anchor)
+        })?;
         write_ms = Some(started.elapsed().as_millis() as u64);
-        crate::logging::info!(
-            "wrote compare views: vrps={} vaps={}",
-            vrps_path.display(),
-            vaps_path.display()
-        );
     }
     Ok(CompareViewTaskOutput { build_ms, write_ms })
 }
@@ -403,6 +402,7 @@ pub(super) fn write_stage_timing(
         && let Some(parent) = path.parent()
     {
         let stage_timing_path = parent.join("stage-timing.json");
+        crate::logging::artifact::write("stage_timing", &stage_timing_path, || {
             std::fs::write(
                 &stage_timing_path,
                 serde_json::to_vec_pretty(stage_timing).map_err(|e| e.to_string())?,
@@ -412,8 +412,8 @@ pub(super) fn write_stage_timing(
                     "write stage timing failed: {}: {e}",
                     stage_timing_path.display()
                 )
+            })
         })?;
-        crate::logging::info!("analysis: wrote {}", stage_timing_path.display());
     }
     Ok(())
 }

src/runtime/post_validation.rs

@@ -40,6 +40,27 @@ fn print_summary_from_shared(validation_time: time::OffsetDateTime, shared: &Pos
         })
     });
     if warning_count > 0 {
+        let mut categories: std::collections::BTreeMap<&str, (usize, Vec<serde_json::Value>)> = std::collections::BTreeMap::new();
+        let warnings = shared.tree_warnings.iter().map(|warning| (warning.category.as_str(), warning.message.as_str(), warning.context.as_deref()))
+            .chain(shared.publication_points.iter().flat_map(|pp| pp.warnings.iter()).map(|warning| (warning.category.as_str(), warning.message.as_str(), warning.context.as_deref())));
+        for (category, message, context) in warnings {
+            let entry = categories.entry(category).or_default();
+            entry.0 += 1;
+            let mut text = crate::logging::redact_text(message);
+            let truncated = text.len() > 1024;
+            let mut boundary = text.len().min(1024);
+            while !text.is_char_boundary(boundary) { boundary -= 1; }
+            text.truncate(boundary);
+            if entry.1.len() < 3 {
+                entry.1.push(serde_json::json!({"message": text, "truncated": truncated}));
+            }
+            crate::logging::emit(crate::logging::Level::Debug, "validation_warning_detail", || serde_json::json!({
+                "category": category, "message": message, "context": context,
+            }));
+        }
+        for (category, (count, examples)) in categories {
+            crate::logging::emit(crate::logging::Level::Info, "validation_warning_category", || serde_json::json!({"category": category, "count": count, "examples": examples}));
+        }
         crate::logging::emit(crate::logging::Level::Warn, "validation_warnings", || {
             serde_json::json!({
                 "count": warning_count, "failed_publication_points": shared.instances_failed,

src/runtime/report_tasks.rs

@@ -80,9 +80,10 @@ pub(super) fn run_ccr_task(
             .map_err(|e| e.to_string())?;
         ccr_build_ms = Some(started.elapsed().as_millis() as u64);
         let started = std::time::Instant::now();
-        write_ccr_file(path, &ccr).map_err(|e| e.to_string())?;
+        crate::logging::artifact::write("ccr", path, || {
+            write_ccr_file(path, &ccr).map_err(|e| e.to_string())
+        })?;
         ccr_write_ms = Some(started.elapsed().as_millis() as u64);
-        crate::logging::info!("wrote CCR: {}", path.display());
     }
 
     Ok(CcrTaskOutput {

src/runtime/run.rs

@@ -1,6 +1,7 @@
 // Top-level CLI execution pipeline.
 
 pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
+    crate::logging::observer::phase("state_store_open");
     let mut policy = read_policy(args.policy_path.as_deref())?;
     if let Some(strict_policy) = args.strict_policy {
         policy.strict = strict_policy;
@@ -109,6 +110,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
         store.as_ref(),
     );
     let validation_started = std::time::Instant::now();
+    crate::logging::observer::phase("runtime_setup");
     let collect_current_repo_objects = false;
     let out = if let Some(dir) = args.rsync_local_dir.as_ref() {
         let http = BlockingHttpFetcher::new(HttpFetcherConfig {
@@ -166,6 +168,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
         || serde_json::json!({"elapsed_ms": validation_ms}),
     );
     let shared = PostValidationShared::from_run_output(out);
+    crate::logging::observer::phase("result_output");
     record_memory_checkpoint(
         &mut memory_checkpoints,
         "after_validation",
@@ -366,8 +369,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
             shared.publication_points.len() as u64,
         );
         let timing_json_path = out_dir.join("timing.json");
-        t.write_json(&timing_json_path, 20)?;
+        crate::logging::artifact::write("timing", &timing_json_path, || t.write_json(&timing_json_path, 20))?;
-        crate::logging::info!("analysis: wrote {}", timing_json_path.display());
     }
 
     // Write the compatibility summary after analysis timing has been flushed.
@@ -376,13 +378,13 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> {
     // point audit, so the final summary can use those authoritative counts.
     if let Some(path) = args.summary_out_path.as_deref() {
         let analysis_counts = timing.as_ref().map(|(_, handle)| handle.counts_snapshot());
-        write_summary(
+        crate::logging::artifact::write("summary", path, || write_summary(
             path,
             &shared,
             &args,
             validation_ms,
             analysis_counts.as_ref(),
-        )?;
+        ))?;
     }
 
     print_summary_from_shared(validation_time, &shared);

src/validation/manifest/models_and_process.rs

@@ -207,6 +207,19 @@ pub enum ManifestFreshError {
     IssuerCaLoadFailed { detail: String },
 }
 
+impl ManifestFreshError {
+    pub(crate) fn warning_category(&self) -> crate::output::report::WarningCategory {
+        use crate::output::report::WarningCategory;
+        match self {
+            Self::Storage(_) | Self::RepoSyncFailed { .. } => WarningCategory::Unclassified,
+            Self::IssuerCaLoadFailed { .. } | Self::EeCertPath(_) | Self::EeResources(_) => WarningCategory::Certificate,
+            Self::ManifestOutsidePublicationPoint { .. } | Self::ManifestNumberNotIncreasing { .. }
+                | Self::ThisUpdateNotIncreasing { .. } | Self::HashMismatch { .. } => WarningCategory::SecurityPolicy,
+            _ => WarningCategory::ManifestCrl,
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum ManifestProcessError {
     #[error("manifest processing failed: {0}")]

src/validation/objects/parallel_stage.rs

@@ -145,6 +145,7 @@ pub(crate) fn prepare_publication_point_for_parallel_roa_and_ta_constraints<
         stats.publication_point_dropped = true;
         warnings.push(
             Warning::new("dropping publication point: no CRL files in validated publication point")
+                .with_category(WarningCategory::ManifestCrl)
                 .with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")])
                 .with_context(manifest_uri),
         );
@@ -246,6 +247,7 @@ pub(crate) fn reduce_parallel_roa_stage(
                     refs.extend_from_slice(extra_rfc_refs_for_crl_selection(&error));
                     warnings.push(
                         Warning::new(format!("dropping invalid ROA: {}: {error}", file.rsync_uri))
+                            .with_category(WarningCategory::ObjectValidation)
                             .with_rfc_refs(&refs)
                             .with_context(&file.rsync_uri),
                     );
@@ -296,6 +298,7 @@ pub(crate) fn reduce_parallel_roa_stage(
                             "dropping invalid ASPA: {}: {error}",
                             file.rsync_uri
                         ))
+                        .with_category(WarningCategory::ObjectValidation)
                         .with_context(&file.rsync_uri),
                     );
                 }

src/validation/objects/serial_processing.rs

@@ -142,6 +142,7 @@ pub fn process_publication_point_for_issuer_with_ta_constraints<P: PublicationPo
         stats.publication_point_dropped = true;
         warnings.push(
             Warning::new("dropping publication point: no CRL files in validated publication point")
+                .with_category(WarningCategory::ManifestCrl)
                 .with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")])
                 .with_context(manifest_uri),
         );

src/validation/run_tree_from_tal/discovery.rs

@@ -271,7 +271,9 @@ fn discover_multiple_roots_from_tal_inputs(
     strict_name: bool,
 ) -> Result<Vec<TalRootDiscovery>, RunTreeFromTalError> {
     let mut roots = Vec::with_capacity(tal_inputs.len());
+    crate::logging::observer::phase("trust_anchor_discovery");
     for tal_input in tal_inputs {
+        crate::logging::emit(crate::logging::Level::Info, "trust_anchor_started", || serde_json::json!({"tal_id": tal_input.tal_id}));
         let discovery = match root_discovery_from_tal_input(
             tal_input,
             http_fetcher,
@@ -292,7 +294,10 @@ fn discover_multiple_roots_from_tal_inputs(
                 );
                 continue;
             }
-            Err(error) => return Err(error.into()),
+            Err(error) => {
+                crate::logging::emit(crate::logging::Level::Error, "trust_anchor_failed", || serde_json::json!({"tal_id": tal_input.tal_id, "error": error.to_string()}));
+                return Err(error.into());
+            }
         };
         let root_handle = root_handle_from_trust_anchor(
             &discovery.trust_anchor,
@@ -305,6 +310,7 @@ fn discover_multiple_roots_from_tal_inputs(
             discovery,
             root_handle,
         });
+        crate::logging::emit(crate::logging::Level::Info, "trust_anchor_completed", || serde_json::json!({"tal_id": tal_input.tal_id}));
     }
     if roots.is_empty() {
         return Err(RunTreeFromTalError::Setup(
@@ -312,6 +318,7 @@ fn discover_multiple_roots_from_tal_inputs(
                 .to_string(),
         ));
     }
+    crate::logging::observer::phase("online_validation");
     Ok(roots)
 }
 

src/validation/tree.rs

@@ -308,6 +308,12 @@ pub fn run_tree_serial_audit_multi_root(
         };
 
         instances_processed += 1;
+        crate::logging::observer::update(instances_processed as u64, || {
+            serde_json::json!({
+                "publication_points_completed": instances_processed,
+                "publication_points_failed": instances_failed,
+            })
+        });
         warnings.extend(res.warnings);
         warnings.extend(res.objects.warnings);
         vrps.extend(res.objects.vrps);

src/validation/tree_parallel/phase2.rs

@@ -164,6 +164,17 @@ pub fn run_tree_parallel_phase2_audit_multi_root(
                     config,
                 );
 
+                crate::logging::observer::update(finished.len() as u64, || serde_json::json!({
+                    "publication_points_completed": finished.len(),
+                    "publication_points_started": instances_started,
+                    "ca_queue": ca_queue.len(), "ready_queue": ready_queue.len(),
+                    "waiting_repository_identities": ca_waiting_repo_by_identity.len(),
+                    "publication_points_inflight": inflight_publication_points.len(),
+                    "pending_object_dispatch": pending_roa_dispatch.len(),
+                    "pending_finalization": pending_finalization.len(),
+                    "finalization_inflight": finalize_inflight,
+                    "staging_inflight": staging_inflight,
+                }));
                 let repo_poll_timeout = event_poll_timeout(
                     &ca_queue,
                     &ready_queue,

src/validation/tree_runner/audit_projection.rs

@@ -127,7 +127,7 @@ fn build_publication_point_audit_from_failed_fetch(
         .map(AuditWarning::from)
         .collect::<Vec<_>>();
     warnings.push(AuditWarning::from(
-        &Warning::new(fresh_error.to_string()).with_context(&ca.manifest_rsync_uri),
+        &Warning::new(fresh_error.to_string()).with_category(fresh_error.warning_category()).with_context(&ca.manifest_rsync_uri),
     ));
     PublicationPointAudit {
         node_id: None,

src/validation/tree_runner/fresh_pipeline.rs

@@ -160,6 +160,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
                 Vec::new(),
                 vec![
                     Warning::new(format!("child CA discovery failed: {e}"))
+                        .with_category(crate::output::report::WarningCategory::Certificate)
                         .with_rfc_refs(&[RfcRef("RFC 6487 §7.2")])
                         .with_context(&ca.manifest_rsync_uri),
                 ],

src/validation/tree_runner/publication_point_runner.rs

@@ -103,6 +103,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
                         "publication point processing failed: {}",
                         stage_error.error
                     ))
+                    .with_category(stage_error.error.warning_category())
                     .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")])
                     .with_context(&ca.manifest_rsync_uri),
                 );

tests/logging_contract.rs

@@ -0,0 +1,70 @@
+use std::process::Command;
+
+#[test]
+fn default_format_is_human_readable_text() {
+    let result = Command::new(env!("CARGO_BIN_EXE_panda-rpki"))
+        .env_remove("PANDA_RPKI_LOG_FORMAT")
+        .env_remove("PANDA_RPKI_LOG_LEVEL")
+        .args(["validate"])
+        .output()
+        .unwrap();
+    let diagnostic = String::from_utf8(result.stderr).unwrap();
+    assert_eq!(result.status.code(), Some(2));
+    assert!(diagnostic.contains("ERROR Command failed"));
+    assert!(!diagnostic.contains("command_failed"));
+    let call_line = include_str!("../src/main.rs")
+        .lines()
+        .position(|line| line.contains("panda_rpki::logging::emit("))
+        .unwrap()
+        + 1;
+    assert!(diagnostic.ends_with(&format!("[src/main.rs:{call_line}]\n")));
+    assert!(result.stdout.is_empty());
+}
+
+#[test]
+fn invalid_progress_interval_is_rejected_without_outputs() {
+    for value in ["-1", "wrong", "18446744073709551616"] {
+        let result = Command::new(env!("CARGO_BIN_EXE_panda-rpki"))
+            .args(["validate", "--log-progress-interval-secs", value])
+            .output()
+            .unwrap();
+        assert_eq!(result.status.code(), Some(2));
+        assert!(result.stdout.is_empty());
+        assert!(String::from_utf8_lossy(&result.stderr).contains("log-progress-interval-secs"));
+    }
+}
+
+#[test]
+fn progress_cli_overrides_invalid_environment() {
+    let result = Command::new(env!("CARGO_BIN_EXE_panda-rpki"))
+        .env("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS", "invalid")
+        .args([
+            "validate",
+            "--log-progress-interval-secs",
+            "0",
+            "--log-format",
+            "json",
+        ])
+        .output()
+        .unwrap();
+    let record: serde_json::Value = serde_json::from_slice(&result.stderr).unwrap();
+    let call_line = include_str!("../src/main.rs")
+        .lines()
+        .position(|line| line.contains("panda_rpki::logging::emit("))
+        .unwrap()
+        + 1;
+    assert_eq!(record["source"]["file"], "src/main.rs");
+    assert_eq!(record["source"]["line"], call_line);
+    assert!(
+        record["fields"]["error"]
+            .as_str()
+            .unwrap()
+            .contains("--tal")
+    );
+    assert!(
+        !record["fields"]["error"]
+            .as_str()
+            .unwrap()
+            .contains("progress")
+    );
+}

tests/support/test_daemon.py

@@ -86,6 +86,16 @@ def run(binary):
             assert gap >= 1.9, gap
             for seq in (1, 2):
                 out = root / f"runs/run_{seq:06}"
+                assert not (out / "stderr.log").exists()
+                assert not (out / "stdout.log").exists()
+                assert (out / "process-output.txt").read_bytes() == b""
+                records = [json.loads(line) for line in (out / "validator.log").read_text().splitlines()]
+                assert records[-1]["event"] == "validation_completed", records[-1]
+                assert all(record["source"]["line"] > 0 and record["source"]["file"].endswith(".rs") for record in records)
+                assert all(record["source"]["file"] not in ("src/logging.rs", "src/logging/progress.rs") for record in records)
+                assert all(record["fields"]["run_seq"] == seq for record in records)
+                assert len({record["fields"]["run_id"] for record in records}) == 1
+                assert any(record["event"] == "artifact_write_completed" for record in records)
                 assert len((out / "vrps.csv").read_text().splitlines()) - 1 == seq
                 subprocess.run(["openssl", "asn1parse", "-inform", "DER", "-in",
                                 str(out / "result.ccr"), "-noout"], check=True)
@@ -106,6 +116,15 @@ def run(binary):
             assert state_is(root, "exited")
 
             # A real validator failure is recorded, retried, and returned as nonzero at max-runs.
+            output_failure = subprocess.run(
+                [binary, "validate", *child_args, "--out", str(base / "bad-output"), "--ccr-out", str(base)],
+                capture_output=True, text=True, timeout=30,
+            )
+            assert output_failure.returncode == 2 and not output_failure.stdout
+            failure_records = [json.loads(line) for line in output_failure.stderr.splitlines()]
+            assert any(record["event"] == "artifact_write_failed" and record["fields"]["artifact_type"] == "ccr" for record in failure_records)
+            assert any(record["event"] == "validation_stage_failed" for record in failure_records)
+            assert not any(record["event"] == "validation_completed" for record in failure_records)
             failed = base / "failed"
             bad_inputs = ["--tal", str(base / "missing.tal"), "--ta", str(base / "missing.cer")]
             failure = launch(failed, ["--max-runs", "2", "--interval-secs", "0"], bad_inputs)
@@ -115,10 +134,33 @@ def run(binary):
 
             # Suspend the HTTPS server so the actual validator cannot complete a request.
             server.send_signal(signal.SIGSTOP)
+            slow_inputs = list(child_args)
+            slow_inputs[slow_inputs.index("--http-timeout-secs") + 1] = "30"
+            slow_inputs += ["--log-progress-interval-secs", "1"]
+            progress_root = base / "progress"
+            progress = launch(progress_root, ["--max-runs", "1", "--run-timeout-secs", "20"], slow_inputs)
+            progress_log = progress_root / "runs/run_000001/validator.log"
+
+            def has_progress():
+                if not progress_log.exists():
+                    return False
+                lines = progress_log.read_text().splitlines()
+                return sum('"event":"validation_progress"' in line for line in lines) >= 2
+
+            wait_for(has_progress, "periodic progress during blocked network", timeout=15)
+            server.send_signal(signal.SIGCONT)
+            assert progress.wait(timeout=30) == 0
+            progress_records = [json.loads(line) for line in progress_log.read_text().splitlines()]
+            assert progress_records[-1]["event"] == "validation_completed"
+            assert any(record["event"] == "validation_progress" and record["fields"]["repository_attempts_inflight"] > 0 for record in progress_records)
+            server.send_signal(signal.SIGSTOP)
             hung = base / "timeout"
-            timeout = launch(hung, ["--max-runs", "1", "--run-timeout-secs", "1"])
+            disabled_inputs = list(slow_inputs)
+            disabled_inputs[-1] = "0"
+            timeout = launch(hung, ["--max-runs", "1", "--run-timeout-secs", "2"], disabled_inputs)
             assert timeout.wait(timeout=10) == 2
             assert read(hung / "runs/run_000001/run-summary.json")["error"] == "timeout"
+            assert '"event":"validation_progress"' not in (hung / "runs/run_000001/validator.log").read_text()
             interrupted = base / "interrupted"
             active_proc = launch(interrupted, ["--shutdown-grace-secs", "0"])
             wait_for(lambda: state_is(interrupted, "running"), "active validator")