Silent-log-source tasks todo list

Everything is relative to the 'silent-log-source' directory!

# Silent Log Source Detector — AI Implementation Spec
> Stack: Python 3.9+ · stdlib + `requests` only · Single file: `silent_log_detector.py`

---

## Assumptions & Clarifications

The following assumptions were made based on the provided design spec (v2) and the existing prototype:

1. **Single-file output**: The final deliverable is one Python script (`silent_log_detector.py`). No package structure, no separate modules. Functions are the unit of decomposition.
2. **Prototype is reference only**: The existing prototype (`silent-log-source/prototype.py`) is a starting point to understand intent, NOT a base to patch. The coding AI rewrites from scratch to implement the full spec. Do not preserve prototype code that contradicts the spec.
3. **Python 3.9+**: `datetime.timezone`, `logging.handlers`, `fnmatch`, `yaml` (via `PyYAML`) are available. `PyYAML` is the only additional dependency beyond `requests` — the spec says "stdlib + requests only" but YAML parsing requires `PyYAML`. Flag this to the user if `PyYAML` is unacceptable; the alternative is a restricted YAML parser in stdlib (`tomllib` in 3.11+ does not parse YAML). **Assumption: `PyYAML` is permitted.**
4. **No tests file**: The spec does not mention a test suite. Acceptance criteria are written to be manually verifiable. The coding AI should not create a separate test file unless instructed.
5. **Removed from prototype, not in spec**: Slack webhook (`--webhook`) is removed entirely.
6. **`detection_run_id`**: Format is `YYYYMMDD-HHMMSS` in UTC, generated once per detection cycle (not per-document).
7. **Log file path**: Not specified in spec. Default to `silent_log_detector.log` in the working directory.

---

## System Frame

**Problem statement:** Detect log sources that have gone silent (stopped sending logs) by comparing their last-seen timestamp from an OpenSearch terms aggregation against a configurable silence threshold, then write alert documents to an OpenSearch index.

**System boundaries:**
- In scope: config loading, exclusion filtering, OpenSearch querying (read), alert document generation, bulk OpenSearch write (write), daemon loop.
- Out of scope: alerting via Slack or email, log parsing, index management, OpenSearch cluster config.

**Architecture:** Single-process procedural Python script with a daemon loop. All state is local to the run cycle. No threads, no queues, no external state.

**Technology choices:**
- Python 3.9+ stdlib + `requests` + `PyYAML`
- OpenSearch REST API (`_search` for detection, `_bulk` for writing)

**Critical constraints:**
- ⚠️ All datetime operations MUST be UTC and timezone-aware (`datetime.timezone.utc`). No naive datetimes anywhere.
- ⚠️ Only `requests`, `PyYAML`, and stdlib are permitted as dependencies.
- ⚠️ The `_bulk` endpoint (NDJSON) must be used for all alert writes — never single-doc `_doc` POST.
- ⚠️ Exclusions use `fnmatch.fnmatch` for glob matching. No regex.
- ⚠️ Config file is reloaded at the top of every cycle when in daemon mode.
- ⚠️ `--insecure` must print a warning to stderr when used. Never silently disable SSL.

**Explicit assumptions:** See section above.

---

## Module Map & Dependency Graph

```
main()
└─► parse_args() # TASK-01
└─► setup_logging() # TASK-02
└─► build_session() # TASK-05
└─► daemon_loop() # TASK-11
└─► load_config() # TASK-03
└─► build_exclusion_set() # TASK-04
└─► run_detection_cycle() # TASK-10
└─► run_index_pass() # TASK-06 (called twice per index: primary + secondary)
└─► consolidate_results() # TASK-07
└─► build_alert_docs() # TASK-08
└─► bulk_write_alerts() # TASK-09
```

**Strict ordering (must build in sequence):**
`TASK-01 → TASK-02 → TASK-03 → TASK-04 → TASK-05 → TASK-06 → TASK-07 → TASK-08 → TASK-09 → TASK-10 → TASK-11 → TASK-12`

No parallel tasks — this is a single file, sequentially implemented.

---

## Project Bootstrap Document

```
# Project Bootstrap — Silent Log Source Detector

## What This Project Does
A production Python script that queries OpenSearch for hosts that have stopped sending logs.
It compares each host's last-seen timestamp against a configurable silence threshold and
writes alert documents to an OpenSearch index. Runs as a one-shot job or a long-running daemon.

## Architecture in One Sentence
Single-file, single-process procedural Python with a config-reload daemon loop; reads from
OpenSearch via _search, writes alerts via _bulk.

## Key Functions (all in silent_log_detector.py)
| Function | Responsibility |
|------------------------|-------------------------------------------------------------|
| parse_args() | CLI parsing — all arguments defined here, nowhere else |
| setup_logging() | Logging config — call once from main(), never elsewhere |
| load_config() | YAML load + defaults inheritance — called every cycle |
| build_exclusion_set() | Glob + expiry filter — returns set of active patterns |
| build_session() | requests.Session with auth + SSL — created once |
| run_index_pass() | Single terms+max agg query against one field |
| consolidate_results() | Merge primary + secondary pass results |
| build_alert_docs() | Produce alert dicts from silent identifiers |
| bulk_write_alerts() | POST NDJSON to _bulk endpoint |
| run_detection_cycle() | Orchestrates one full scan across all configured indexes |
| daemon_loop() | Reload config, run cycle, sleep, repeat |
| main() | Entry point — wires everything together |

## System Invariants
1. All datetime objects are UTC and timezone-aware — no naive datetimes, ever.
2. Exclusions are always evaluated via fnmatch.fnmatch — no string equality, no regex.
3. The _bulk endpoint is always used for writes — never _doc POST.
4. Config is reloaded at the top of every daemon loop iteration, before any query runs.
5. --insecure always prints a WARNING to stderr before disabling SSL verification.

## Out of Scope (Do Not Add)
- Slack or email alerting (removed from spec)
- Multi-file project structure (this is one file)
- Threading or async
- Any dependency not in: requests, PyYAML, Python stdlib
```

---

## Task List

| Task | Name | Depends On |
|------|------|-----------|
| TASK-01 | CLI Parser | — |
| TASK-02 | Logging Subsystem | TASK-01 |
| TASK-03 | Config Loader & Inheritance | TASK-02 |
| TASK-04 | Exclusion Engine | TASK-03 |
| TASK-05 | OpenSearch Session Builder | TASK-01 |
| TASK-06 | Query Engine (Single Pass) | TASK-05 |
| TASK-07 | Dual-Field Consolidation | TASK-06 |
| TASK-08 | Alert Document Builder | TASK-07 |
| TASK-09 | Bulk Indexer | TASK-08, TASK-05 |
| TASK-10 | Per-Index Detection Orchestrator | TASK-04 through TASK-09 |
| TASK-11 | Daemon Loop | TASK-10 |
| TASK-12 | main() Entry Point & Top-Level Wiring | TASK-11 |

---

## AI Instruction Packets

---

### TASK-01: CLI Parser

**Objective:** Implement `parse_args() -> argparse.Namespace` that parses all CLI arguments exactly as specified and returns a `Namespace` object used by every other function.

**Bootstrap Context:**
This is the first function in the file. Read only the CLI section (Section 5) of the design spec.
Key facts:
- `--opensearch-url` is REQUIRED (no default).
- `--sleep` accepts human durations like `24h`, `12h`, `30m`, or raw integers (seconds). Parse these here, store as integer seconds in `args.sleep_seconds`. This overrides `sleep_after_run_seconds` from config.
- `--verbose` / `-v` sets log level to DEBUG.
- `--version` prints a version string and exits.
Stop after implementing `parse_args()`. Do not write any other functions yet.

**Files to Create / Modify:**
- `silent_log_detector.py` — CREATE — Write the file with imports and `parse_args()` only.

**Inputs:**
- `sys.argv` (consumed automatically by argparse)

**Outputs:**
- `argparse.Namespace` with these attributes:
```
args.config # str, default="./silent-log-detector.yaml"
args.opensearch_url # str, required
args.user # str or None
args.password # str or None (note: --pass is reserved; use --password in argparse, expose as --pass via dest)
args.ca_cert # str or None
args.insecure # bool, default=False
args.sleep_seconds # int or None (None = not specified via CLI; config value used instead)
args.dry_run # bool, default=False
args.verbose # bool, default=False
args.version # handled by argparse action='version'
```

**Interface Contract:**
```python
def parse_args() -> argparse.Namespace:
...
```
The `--pass` flag must use `dest='password'` so Python does not complain about the leading dash.
Human duration parsing for `--sleep`:
- `"24h"` → `86400`
- `"12h"` → `43200`
- `"30m"` → `1800`
- `"3600"` → `3600` (raw int string)
- Any unrecognised pattern → `argparse.ArgumentTypeError`

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- `--opensearch-url` MUST be `required=True`. The script must refuse to start without it.
- `--pass` MUST be implemented as `parser.add_argument('--pass', dest='password', ...)`. Do not use `--password` as the CLI flag name (it conflicts with user expectation from the spec).
- Human duration parsing MUST happen inside an `argparse` type function, not in `main()`.
- Do NOT strip trailing slashes from `--opensearch-url` here. That happens in TASK-05.

**Must NOT do:**
- Do not implement any other function in this task.
- Do not read the config file.
- Do not set up logging.

**Acceptance Criteria:**
- [ ] `--opensearch-url` omitted → argparse prints error and exits with code 2.
- [ ] `--sleep 24h` sets `args.sleep_seconds = 86400`.
- [ ] `--sleep 30m` sets `args.sleep_seconds = 1800`.
- [ ] `--sleep 3600` sets `args.sleep_seconds = 3600`.
- [ ] `--sleep badvalue` → argparse error, exit code 2.
- [ ] `--sleep` omitted → `args.sleep_seconds = None`.
- [ ] `--pass secret` sets `args.password = "secret"`.
- [ ] `--verbose` sets `args.verbose = True`.
- [ ] `--dry-run` sets `args.dry_run = True`.
- [ ] `--insecure` sets `args.insecure = True`.

**Edge Cases to Handle:**
- `--sleep 0` → valid, parse as integer `0` (means run-once).
- `--sleep 0h` → valid, parse as `0`.

**Test Requirements:**
Manually invoke `parse_args()` by setting `sys.argv` before calling, or use `parser.parse_args([...])`. Verify all acceptance criteria by inspection.

**Known Risks / Likely Mistakes:**
- AI may name the flag `--password` instead of `--pass` → use `add_argument('--pass', dest='password')`.
- AI may put human duration parsing inline in `main()` instead of as an `argparse` type → put it in a helper `_parse_sleep_duration(value: str) -> int` called as `type=_parse_sleep_duration` in `add_argument`.

---

### TASK-02: Logging Subsystem

**Objective:** Implement `setup_logging(verbose: bool, log_file: str = "silent_log_detector.log") -> None` that configures the stdlib `logging` module with UTC timestamps, optional ANSI console colours (tty-only), and a size-based rotating file handler.

**Bootstrap Context:**
TASK-01 must be complete. Read only Section 6 (Logging) of the design spec.
Key facts:
- UTC timestamps in format: `2026-03-24 09:15:32 UTC [INFO] message`
- ANSI colours: only when `sys.stdout.isatty()` is True. Never colour the file handler.
- `RotatingFileHandler`: 10 MB max, 10 backups.
- `--verbose` → DEBUG level on root logger. Default → INFO level.
Stop after implementing `setup_logging()`. Do not write other functions.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `setup_logging()` and a `UTCFormatter` class below `parse_args()`.

**Inputs:**
- `verbose: bool` — if True, set root logger to DEBUG; else INFO.
- `log_file: str` — path to rotating log file.

**Outputs:**
- Root logger is configured. All subsequent `logging.info/debug/warning/error` calls produce correct output.

**Interface Contract:**
```python
class UTCFormatter(logging.Formatter):
"""Formats log records with UTC timestamps."""
converter = time.gmtime # Force UTC

def setup_logging(verbose: bool, log_file: str = "silent_log_detector.log") -> None:
...
```

Log format string (exact):
```
%(asctime)s UTC [%(levelname)s] %(message)s
```
`asctime` format: `%Y-%m-%d %H:%M:%S`

ANSI colour mapping (console handler only, tty-only):
```
DEBUG → \033[36m (cyan)
INFO → \033[0m (reset/white)
WARNING → \033[33m (yellow)
ERROR → \033[31m (red)
```
Each log line should apply the colour prefix before levelname and reset after the message.

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- The file handler MUST NEVER have ANSI codes — file output must be plain text.
- `UTCFormatter.converter = time.gmtime` — this is how stdlib logging produces UTC timestamps. Do not use `datetime.utcnow()`.
- ANSI colours MUST be conditional on `sys.stdout.isatty()`. In a pipe/container, no colour.
- Do NOT call `logging.basicConfig()` — configure handlers manually to ensure file and console handlers use different formatters.

**Must NOT do:**
- Do not add a third-party logging library.
- Do not modify `parse_args()`.

**Acceptance Criteria:**
- [ ] Log lines contain `UTC` in the timestamp string (e.g. `2026-03-24 09:15:32 UTC [INFO]`).
- [ ] `verbose=True` → DEBUG messages appear on console.
- [ ] `verbose=False` → DEBUG messages do NOT appear.
- [ ] `log_file` exists on disk after `setup_logging()` is called (even if no messages logged yet — RotatingFileHandler creates the file).
- [ ] File handler output contains no ANSI escape codes.
- [ ] Console handler only emits ANSI codes when `sys.stdout.isatty()` returns True.

**Edge Cases to Handle:**
- Log directory does not exist → let `RotatingFileHandler` raise `FileNotFoundError` naturally; do not silently create directories.
- `log_file=""` → fall back to `"silent_log_detector.log"`.

**Known Risks / Likely Mistakes:**
- AI may use `logging.basicConfig()` → this prevents adding a second handler with different formatter. Use `logging.getLogger()` + `addHandler()` explicitly.
- AI may forget `UTCFormatter.converter = time.gmtime` and produce local time → the `converter` class attribute is the only correct approach.
- AI may apply ANSI codes to the file handler → check `sys.stdout.isatty()` and create two separate `Formatter` instances.

---

### TASK-03: Config Loader & Defaults Inheritance

**Objective:** Implement `load_config(config_path: str) -> dict` that reads the YAML config file, applies defaults inheritance to every index entry, loads and merges the optional `exclusions_file`, and returns a fully-resolved config dict.

**Bootstrap Context:**
TASK-01 and TASK-02 must be complete. Read Section 2 (Configuration File) of the design spec.
Key facts:
- Default keys: `host_field`, `secondary_host_field`, `baseline_days`, `silence_hours`, `exclusions`.
- Per-index values REPLACE (not merge) the corresponding default. Exception: if the per-index `exclusions` key is absent, the index inherits the global `defaults.exclusions`.
- `exclusions_file` is optional. If present, its lines are appended to the resolved exclusions list for every index. Lines beginning with `#` are comments. Lines may have `expires:YYYY-MM-DD` suffix (whitespace-separated).
- `sleep_after_run_seconds: 0` means run-once; `> 0` means daemon. This key lives at the top level of the config.
Stop after implementing `load_config()` and its helpers. Do not write query or session functions.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `load_config()` and `_parse_exclusions_file()` helper below `setup_logging()`.

**Inputs:**
- `config_path: str` — path to YAML file.

**Outputs:**
- A `dict` with this exact resolved structure:
```python
{
"environment": str,
"tags": list[str],
"sleep_after_run_seconds": int,
"alerts": {
"opensearch_index": str | None
},
"indexes": [
{
"name": str, # e.g. "logs-*"
"host_field": str, # resolved (from index or defaults)
"secondary_host_field": str | None, # resolved
"baseline_days": int, # resolved
"silence_hours": int | float, # resolved
"exclusions": [ # resolved: inline + exclusions_file lines
{
"pattern": str,
"expires": datetime.date | None # None if no expiry
}
]
},
...
]
}
```

**Interface Contract:**
```python
def load_config(config_path: str) -> dict:
"""Load, resolve, and return fully-inherited config. Raises FileNotFoundError or
yaml.YAMLError on bad input. Logs inheritance at DEBUG level."""

def _parse_exclusions_file(filepath: str) -> list[dict]:
"""Parse an exclusions file. Returns list of {"pattern": str, "expires": date|None}.
Lines starting with # are skipped. Malformed expires: values are logged as WARNING
and treated as no expiry."""
```

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- Per-index `exclusions` key, if present, COMPLETELY REPLACES `defaults.exclusions`. They are NOT merged. If absent, the index inherits `defaults.exclusions`.
- `secondary_host_field` defaults to `null` (Python `None`) if not set at defaults or index level.
- `expires` dates in the exclusions list MUST be `datetime.date` objects, not strings, when returned.
- `exclusions_file` lines are appended AFTER inline exclusions in the resolved list for every index.
- Log a DEBUG message for each inherited field: `"Inheriting silence_hours=24 from defaults for index logs-*"`.
- `alerts.opensearch_index` defaults to `None` if the `alerts` block is absent.

**Must NOT do:**
- Do not filter expired exclusions here — that is TASK-04's job.
- Do not validate OpenSearch connectivity.
- Do not modify `parse_args()` or `setup_logging()`.

**Acceptance Criteria:**
- [ ] Config with no per-index overrides → all index entries have defaults' values.
- [ ] Config with `silence_hours: 48` on one index → that index has `silence_hours=48`, others have the default.
- [ ] `exclusions_file` with `old-relay-* expires:2026-04-15` → parsed as `{"pattern": "old-relay-*", "expires": datetime.date(2026, 4, 15)}`.
- [ ] `exclusions_file` absent or `null` → no error; exclusions are only inline.
- [ ] Index with no `exclusions` key → inherits `defaults.exclusions` patterns.
- [ ] Index with explicit `exclusions: []` → empty exclusions list (does NOT inherit defaults).
- [ ] `tags` absent from YAML → `config["tags"]` is `[]` (empty list, not an error).
- [ ] `FileNotFoundError` raised when config path does not exist.
- [ ] `yaml.YAMLError` propagates when YAML is malformed.

**Edge Cases to Handle:**
- `exclusions_file` line with no pattern (blank after stripping comment) → skip silently.
- `expires:` value that is not a valid ISO date → log WARNING, set `expires=None`.
- `secondary_host_field` explicitly set to `null` in YAML → stored as Python `None`.

**Known Risks / Likely Mistakes:**
- AI may merge exclusion lists instead of replacing → re-read the spec: per-index `exclusions` REPLACES defaults.
- AI may store `expires` as a string instead of `datetime.date` → parse with `datetime.date.fromisoformat()`.
- AI may silently swallow `yaml.YAMLError` → let it propagate to the caller.

---

### TASK-04: Exclusion Engine

**Objective:** Implement `build_exclusion_set(raw_exclusions: list[dict]) -> list[str]` that filters expired exclusion entries and returns a list of active glob patterns, and `is_excluded(identifier: str, patterns: list[str]) -> bool` that checks a host identifier against those patterns.

**Bootstrap Context:**
TASK-03 must be complete. Read Section 2 exclusions rules and Section 3 runtime behaviour from the design spec.
Key facts:
- Exclusion entries are `{"pattern": str, "expires": datetime.date | None}` — exactly as produced by `load_config()`.
- An exclusion is active if: `expires is None` OR `expires >= today (UTC)`.
- Glob matching uses `fnmatch.fnmatch(identifier, pattern)`.
- `is_excluded` returns `True` if the identifier matches ANY active pattern.
Stop after implementing these two functions.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `build_exclusion_set()` and `is_excluded()` below `load_config()`.

**Inputs (build_exclusion_set):**
- `raw_exclusions: list[dict]` — list of `{"pattern": str, "expires": date | None}` from `load_config()`.

**Inputs (is_excluded):**
- `identifier: str` — the hostname or IP string to test.
- `patterns: list[str]` — active patterns from `build_exclusion_set()`.

**Interface Contract:**
```python
def build_exclusion_set(raw_exclusions: list[dict]) -> list[str]:
"""Filter expired entries. Return list of active glob pattern strings.
Expiry comparison uses UTC today. Logs count of expired patterns filtered at DEBUG."""

def is_excluded(identifier: str, patterns: list[str]) -> bool:
"""Return True if identifier matches any pattern via fnmatch.fnmatch."""
```

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- Use `fnmatch.fnmatch` — NOT `re.match`, NOT `in`, NOT `str.startswith`.
- "Today" for expiry comparison MUST be `datetime.datetime.now(datetime.timezone.utc).date()`.
- An entry with `expires=None` is ALWAYS active (never expired).
- An entry with `expires = today` is STILL ACTIVE — expiry is exclusive on the day after.

**Must NOT do:**
- Do not modify TASK-03 functions.
- Do not load the config file here — inputs come from `load_config()`.

**Acceptance Criteria:**
- [ ] Pattern `"*.staging.example.com"` → `is_excluded("db01.staging.example.com", ...)` returns `True`.
- [ ] Pattern `"test-*"` → `is_excluded("test-server-01", ...)` returns `True`.
- [ ] Pattern `"test-*"` → `is_excluded("prod-server-01", ...)` returns `False`.
- [ ] Entry with `expires = yesterday` → filtered out (not in active set).
- [ ] Entry with `expires = today` → kept (still active).
- [ ] Entry with `expires = None` → always kept.
- [ ] Empty `raw_exclusions` list → `build_exclusion_set` returns `[]`.

**Edge Cases to Handle:**
- Empty `identifier` string → `fnmatch.fnmatch("", pattern)` — let fnmatch handle it naturally.
- `patterns` is empty list → `is_excluded` returns `False`.

**Known Risks / Likely Mistakes:**
- AI may use `datetime.date.today()` (system local time) instead of UTC today → must use `datetime.datetime.now(datetime.timezone.utc).date()`.
- AI may use `expires > today` (strict greater-than) making today's entries expire early → must use `>=`.

---

### TASK-05: OpenSearch Session Builder

**Objective:** Implement `build_session(args: argparse.Namespace) -> tuple[requests.Session, str]` that creates an authenticated `requests.Session` with SSL configured, resolves credentials via the auth precedence chain, and returns the session and the cleaned base URL.

**Bootstrap Context:**
TASK-01 must be complete. Read Section 4 (Authentication & Connection) of the design spec.
Key facts:
- Auth precedence: CLI `--user`/`--pass` → env `OPENSEARCH_USER`/`OPENSEARCH_PASSWORD` → interactive `getpass.getpass()`.
- SSL: `--ca-cert` path → `session.verify = ca_cert_path`. `--insecure` → `session.verify = False` + print WARNING to `sys.stderr`.
- Strip trailing slash from URL. Return the cleaned URL as second return value.
- Set `Content-Type: application/json` header on session.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `build_session()` below the exclusion engine.

**Inputs:**
- `args: argparse.Namespace` — as produced by `parse_args()`.

**Outputs:**
- `(session, base_url)` where `session` is a configured `requests.Session` and `base_url` is the cleaned URL string.

**Interface Contract:**
```python
def build_session(args: argparse.Namespace) -> tuple[requests.Session, str]:
"""
Resolve credentials (CLI → env → prompt), configure SSL, return (session, base_url).
Prints a WARNING to sys.stderr if --insecure is used.
Prompts interactively only if both CLI and env credentials are absent.
"""
```

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- `--insecure` MUST print to `sys.stderr`: `"WARNING: SSL verification disabled. Connection is not secure."` Then set `session.verify = False` AND `requests.packages.urllib3.disable_warnings()`.
- Never log the password value — not at DEBUG level, not anywhere.
- Trailing slash on URL MUST be stripped: `base_url = args.opensearch_url.rstrip('/')`.
- If both `--ca-cert` and `--insecure` are supplied, `--insecure` wins (more permissive), log WARNING.

**Must NOT do:**
- Do not write queries here.
- Do not read the config file.

**Acceptance Criteria:**
- [ ] `--user alice --pass secret` → `session.auth == ("alice", "secret")`.
- [ ] `--user` absent, env `OPENSEARCH_USER=bob`, `OPENSEARCH_PASSWORD=pass` set → `session.auth == ("bob", "pass")` without prompting.
- [ ] Neither CLI nor env → `getpass.getpass()` is called for password, `input()` for username.
- [ ] `--ca-cert /path/to/ca.pem` → `session.verify == "/path/to/ca.pem"`.
- [ ] `--insecure` → `session.verify == False` and stderr contains `"WARNING: SSL verification disabled"`.
- [ ] `"https://opensearch.example.com/"` → returned base URL is `"https://opensearch.example.com"` (no trailing slash).

**Edge Cases to Handle:**
- `OPENSEARCH_PASSWORD` is set but `OPENSEARCH_USER` is not → treat as if neither env var is set, fall through to prompt.
- Username prompted interactively → use `input("OpenSearch username: ")`.

**Known Risks / Likely Mistakes:**
- AI may check env vars individually instead of requiring both → must require both `OPENSEARCH_USER` AND `OPENSEARCH_PASSWORD` to be non-empty for the env branch to activate.
- AI may log credentials at DEBUG level → never log password.

---

### TASK-06: Query Engine — Single Detection Pass

**Objective:** Implement `run_index_pass(session: requests.Session, base_url: str, index_name: str, host_field: str, baseline_days: int, agg_size: int = 10000) -> dict[str, str]` that queries OpenSearch for all unique values of `host_field` seen in the baseline window and returns their last-seen timestamps.

**Bootstrap Context:**
TASK-05 must be complete. Read Section 1 (High-Level Purpose) and Section 10 (Production Details).
Key facts:
- Query uses a terms aggregation (size 10,000) + inner max aggregation on `@timestamp`.
- Baseline window: `gte: now-{baseline_days}d/d, lt: now/d`.
- Result: mapping of `{identifier_value: iso_timestamp_string}`.
- The `@timestamp` returned from a `max` aggregation includes a `value_as_string` key (ISO format with Z).
- This function is called twice per index: once for `host_field`, once for `secondary_host_field`.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `run_index_pass()` below `build_session()`.

**Inputs:**
- `session`: authenticated `requests.Session`.
- `base_url`: cleaned base URL string (no trailing slash).
- `index_name`: e.g. `"logs-*"`.
- `host_field`: e.g. `"host.hostname.keyword"` or `"host.ip"`.
- `baseline_days`: integer.
- `agg_size`: integer, default 10000.

**Outputs:**
```python
dict[str, str]
# e.g. {"db-01.prod": "2026-03-23T08:12:11.000Z", "192.168.1.5": "2026-03-22T10:00:00.000Z"}
# Returns {} on empty results or query error (error is logged, not raised).
```

**Interface Contract:**
```python
def run_index_pass(
session: requests.Session,
base_url: str,
index_name: str,
host_field: str,
baseline_days: int,
agg_size: int = 10000
) -> dict[str, str]:
"""
Run terms+max aggregation. Returns {identifier: last_seen_iso_string}.
On HTTP error or exception: log ERROR and return {}.
"""
```

Query body:
```json
{
"size": 0,
"query": {
"range": {
"@timestamp": {
"gte": "now-{baseline_days}d/d",
"lt": "now/d"
}
}
},
"aggs": {
"by_host": {
"terms": {
"field": "{host_field}",
"size": {agg_size}
},
"aggs": {
"last_seen": {
"max": {"field": "@timestamp"}
}
}
}
}
}
```

URL pattern: `POST {base_url}/{index_name}/_search`
Use `urllib.parse.quote(index_name, safe='')` to encode the index name in the URL.

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- MUST use `requests.Session.post()` — not `requests.post()`.
- On ANY non-200 response: log `ERROR` with status code and response body, return `{}`. Do NOT raise an exception.
- On ANY exception (`requests.RequestException`, etc.): log `ERROR`, return `{}`. Do NOT crash.
- `value_as_string` is the timestamp field. Check it exists; if a bucket has no `value_as_string`, skip that bucket.

**Must NOT do:**
- Do not filter silent hosts here — this function only retrieves raw last-seen data.
- Do not apply exclusions.
- Do not call this function directly — TASK-10 calls it.

**Acceptance Criteria:**
- [ ] Returns `{}` (not an exception) when OpenSearch returns HTTP 400 or 500.
- [ ] Returns `{}` (not an exception) on `requests.ConnectionError`.
- [ ] Returns correct `{hostname: iso_timestamp}` dict when aggregation succeeds.
- [ ] Index name containing `*` (e.g. `"logs-*"`) is URL-encoded correctly in the request path.
- [ ] Logs DEBUG message with the count of hosts returned: `"Loaded 1243 hosts from primary pass on logs-*"`.

**Edge Cases to Handle:**
- Aggregation returns zero buckets → return `{}`.
- Bucket exists but `last_seen.value_as_string` is absent (null max) → skip that bucket, log DEBUG.

**Known Risks / Likely Mistakes:**
- AI may use `requests.post()` instead of `session.post()` → loses auth headers.
- AI may raise on HTTP errors instead of returning `{}` → must swallow and log.
- AI may forget to URL-encode the index name → `logs-*` must become `logs-%2A` in the URL path via `quote(index_name, safe='')`.

---

### TASK-07: Dual-Field Consolidation

**Objective:** Implement `consolidate_results(primary: dict, secondary: dict, silence_hours: float, index_name: str) -> list[dict]` that merges two detection passes, deduplicates by identifier string, selects the most-recent `last_seen` when a key appears in both, computes `hours_silent`, and returns only identifiers that exceed the silence threshold.

**Bootstrap Context:**
TASK-06 must be complete. Read Section 7 (Dual / Fallback Host Field Support) of the design spec carefully.
Key facts:
- `primary` and `secondary` are both `dict[str, str]` (identifier → iso_timestamp) from `run_index_pass()`.
- If the same string key appears in both dicts, keep the MORE RECENT `last_seen` (max of the two timestamps).
- Each result entry records `identifier_type`: `"primary"` if the identifier came from the primary pass only or won the merge, `"secondary"` if it came from secondary only.
- Only identifiers where `hours_silent >= silence_hours` are returned.
- `hours_silent` is computed against UTC now.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `consolidate_results()` below `run_index_pass()`.

**Inputs:**
- `primary: dict[str, str]` — `{identifier: iso_timestamp}` from primary host field pass.
- `secondary: dict[str, str]` — `{identifier: iso_timestamp}` from secondary pass (may be `{}` if no secondary field configured).
- `silence_hours: float` — threshold (e.g. `24.0`).
- `index_name: str` — for logging only.

**Outputs:**
```python
list[dict]
# Each entry:
{
"identifier": str, # the hostname or IP string
"last_seen": str, # ISO timestamp string
"hours_silent": float, # rounded to 1 decimal
"identifier_type": "primary" | "secondary", # string literal
"host_field_used": str # NOT set here — set in TASK-10 when calling this function
}
```
Note: `host_field_used` is NOT populated by this function — it is added in TASK-10.

**Interface Contract:**
```python
def consolidate_results(
primary: dict[str, str],
secondary: dict[str, str],
silence_hours: float,
index_name: str,
now: datetime.datetime # pass UTC now explicitly for testability
) -> list[dict]:
"""
Merge primary + secondary, deduplicate, filter by silence threshold.
Returns list of silent identifier dicts (without host_field_used).
"""
```
Timestamp comparison: parse ISO strings to `datetime.datetime` objects using:
`datetime.datetime.fromisoformat(ts.replace('Z', '+00:00'))`

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- `now` MUST be passed in as a parameter (not computed inside the function) to ensure testability.
- When the same key exists in both `primary` and `secondary`, the identifier_type is `"primary"` if primary's timestamp ≥ secondary's, `"secondary"` otherwise.
- `hours_silent` MUST be `round((now - last_seen_dt).total_seconds() / 3600, 1)`.
- Only identifiers with `hours_silent >= silence_hours` appear in output.

**Must NOT do:**
- Do not apply exclusions here — TASK-10 applies them before calling this function.
- Do not write alert documents — that is TASK-08.
- Do not call OpenSearch.

**Acceptance Criteria:**
- [ ] Key exists only in `primary` → `identifier_type == "primary"`.
- [ ] Key exists only in `secondary` → `identifier_type == "secondary"`.
- [ ] Key in both, primary timestamp newer → `identifier_type == "primary"`, last_seen = primary timestamp.
- [ ] Key in both, secondary timestamp newer → `identifier_type == "secondary"`, last_seen = secondary timestamp.
- [ ] Identifier with `hours_silent = 23.9`, `silence_hours = 24` → NOT in output.
- [ ] Identifier with `hours_silent = 24.0`, `silence_hours = 24` → IN output.
- [ ] Empty `primary` and empty `secondary` → returns `[]`.

**Edge Cases to Handle:**
- Malformed ISO timestamp in input dict → log WARNING, skip that identifier.
- `secondary` is `{}` (secondary field not configured) → function handles gracefully, returns only primary results.

**Known Risks / Likely Mistakes:**
- AI may compute `now` inside the function instead of accepting it as a parameter → breaks testability.
- AI may use `>` instead of `>=` for the silence threshold → `hours_silent == silence_hours` must be included.

---

### TASK-08: Alert Document Builder

**Objective:** Implement `build_alert_doc(entry: dict, index_config: dict, config: dict, detection_run_id: str, primary_host_field: str, secondary_host_field: str | None) -> dict` that constructs the exact alert document schema defined in the spec.

**Bootstrap Context:**
TASK-07 must be complete. Read Section 8 (Alert Document) of the design spec.
Key facts:
- The alert document has exactly these fields — no more, no fewer.
- `silence_hours` in the alert doc is the computed `hours_silent` from the silent identifier entry (i.e. how long it has actually been silent), NOT the threshold.
- `host_field` in the alert is the actual field used for this identifier (primary or secondary field string).
- `hostname` field holds the identifier value regardless of whether it's a hostname or IP.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `build_alert_doc()` below `consolidate_results()`.

**Inputs:**
- `entry: dict` — one item from `consolidate_results()` output.
- `index_config: dict` — the resolved index config entry from `load_config()`.
- `config: dict` — the full top-level config dict (for `environment` and `tags`).
- `detection_run_id: str` — format `"YYYYMMDD-HHMMSS"` (UTC), generated once per cycle.
- `primary_host_field: str` — the `host_field` string from index config.
- `secondary_host_field: str | None` — the `secondary_host_field` string from index config (or None).

**Interface Contract:**
```python
def build_alert_doc(
entry: dict,
index_config: dict,
config: dict,
detection_run_id: str,
primary_host_field: str,
secondary_host_field: str | None,
now: datetime.datetime
) -> dict:
```

**Output — exact document shape (spec Section 8):**
```json
{
"@timestamp": "",
"hostname": "",
"silence_hours": "",
"last_seen": "",
"source_index_pattern": "",
"host_field": "",
"identifier_type": "",
"environment": "",
"tags": [""],
"alert_type": "silent_log_source",
"detection_run_id": ""
}
```

`@timestamp` format: `now.strftime('%Y-%m-%dT%H:%M:%SZ')`

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- `host_field` in the alert document MUST reflect which field was actually used: `primary_host_field` when `identifier_type == "primary"`, `secondary_host_field` when `"secondary"`.
- `silence_hours` in the doc is `entry['hours_silent']` (the actual elapsed hours), not the config threshold.
- Do NOT add any extra fields beyond those listed. Other tasks/consumers depend on the exact schema.
- `tags` must be a list even if the config has a single tag.

**Must NOT do:**
- Do not call OpenSearch.
- Do not apply exclusions.
- Do not modify the `entry` dict in place — return a new dict.

**Acceptance Criteria:**
- [ ] `identifier_type == "primary"` → `host_field` equals `primary_host_field`.
- [ ] `identifier_type == "secondary"` → `host_field` equals `secondary_host_field`.
- [ ] `@timestamp` ends with `Z` and is in ISO 8601 format.
- [ ] `silence_hours` equals `entry['hours_silent']`, not `index_config['silence_hours']`.
- [ ] `alert_type` is the literal string `"silent_log_source"`.
- [ ] Output dict has exactly the 12 fields listed above — no more.

**Edge Cases to Handle:**
- `secondary_host_field` is None but `identifier_type == "secondary"` → this should never happen (TASK-07 constraint), but if it does: set `host_field` to `""` and log ERROR.

**Known Risks / Likely Mistakes:**
- AI may put the config's `silence_hours` threshold instead of the actual `hours_silent` → use `entry['hours_silent']`.
- AI may add extra fields ("helpful" ones like `hours_threshold`) → do not add anything not in the schema.

---

### TASK-09: Bulk Alert Indexer

**Objective:** Implement `bulk_write_alerts(session: requests.Session, base_url: str, alert_index: str, alert_docs: list[dict], dry_run: bool) -> None` that sends all alert documents to OpenSearch via the `_bulk` endpoint in NDJSON format, in a single HTTP POST.

**Bootstrap Context:**
TASK-05 and TASK-08 must be complete. Read Section 9 (Bulk Indexing) of the design spec.
Key facts:
- NDJSON format: alternating action line + document line, each terminated by `\n`.
- Action line: `{"index": {"_index": ""}}`
- One HTTP POST to `{base_url}/_bulk`.
- Check the bulk response for per-document errors in `response_body["items"]`.
- `dry_run=True` → log each document at INFO, do NOT POST.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `bulk_write_alerts()` below `build_alert_doc()`.

**Inputs:**
- `session`: authenticated `requests.Session`.
- `base_url`: cleaned base URL (no trailing slash).
- `alert_index`: e.g. `"silent-log-sources-alerts"`.
- `alert_docs`: list of alert dicts from `build_alert_doc()`.
- `dry_run`: if True, skip the POST.

**Interface Contract:**
```python
def bulk_write_alerts(
session: requests.Session,
base_url: str,
alert_index: str,
alert_docs: list[dict],
dry_run: bool
) -> None:
"""
Build NDJSON bulk payload, POST to _bulk. Log errors per document.
If dry_run=True, log documents and return without posting.
If alert_docs is empty, return immediately without posting.
"""
```

NDJSON construction:
```python
lines = []
for doc in alert_docs:
lines.append(json.dumps({"index": {"_index": alert_index}}))
lines.append(json.dumps(doc))
payload = "\n".join(lines) + "\n"
```

POST to: `{base_url}/_bulk`
Headers: `Content-Type: application/x-ndjson` (override session default for this call only).

Bulk response error checking:
```python
body = response.json()
if body.get("errors"):
for item in body.get("items", []):
op = item.get("index", {})
if op.get("status", 0) >= 400:
logging.error(f"Bulk index error for doc: {op.get('error')}")
```

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- Content-Type for the bulk POST MUST be `application/x-ndjson`, NOT `application/json`.
- The payload MUST end with a trailing newline `\n`.
- Individual document errors MUST NOT stop the overall run — log ERROR and continue.
- `dry_run=True` MUST log each doc but MUST NOT POST.
- If `alert_docs` is empty, return immediately with no POST and no error.

**Must NOT do:**
- Do not POST individual documents — all must go in one bulk request.
- Do not modify alert documents.
- Do not modify session headers permanently — override `Content-Type` for this single call using `session.post(url, data=payload, headers={"Content-Type": "application/x-ndjson"})`.

**Acceptance Criteria:**
- [ ] Empty `alert_docs` → no HTTP call made.
- [ ] `dry_run=True` with non-empty docs → logs docs, no HTTP call.
- [ ] NDJSON payload has `2 * len(alert_docs)` lines plus trailing newline.
- [ ] `Content-Type: application/x-ndjson` is sent (not `application/json`).
- [ ] Bulk response with `errors: true` → each failed item logged at ERROR, function does not raise.
- [ ] POST to `{base_url}/_bulk` (not `{base_url}/{index}/_bulk`).

**Edge Cases to Handle:**
- `response.json()` fails to parse → log ERROR with raw response text, return.
- HTTP 429 or 503 from bulk endpoint → log ERROR, return. No retry (not in spec).

**Known Risks / Likely Mistakes:**
- AI may POST to `{base_url}/{alert_index}/_bulk` → must use the root `/_bulk` endpoint.
- AI may use `application/json` content type → must be `application/x-ndjson`.
- AI may permanently mutate `session.headers` → use per-request headers override.

---

### TASK-10: Per-Index Detection Orchestrator

**Objective:** Implement `run_detection_cycle(session: requests.Session, base_url: str, config: dict, active_patterns: dict[str, list[str]], args: argparse.Namespace) -> list[dict]` that iterates all configured indexes, runs dual-pass detection, applies exclusions, and returns the full list of alert documents ready for bulk write.

**Bootstrap Context:**
TASK-04 through TASK-09 must be complete. Read Sections 7, 8, and 10 of the design spec.
Key facts:
- `active_patterns` is a dict keyed by index name, mapping to the active exclusion pattern list for that index (pre-computed by TASK-04 for each index).
- For each index: run primary pass, run secondary pass (if `secondary_host_field` is not None), call `consolidate_results()`, apply exclusions via `is_excluded()`, build alert docs.
- Generate `detection_run_id` once at the start of this function (not per index).
- `now` is computed once at the start and passed to `consolidate_results()` and `build_alert_doc()`.
- `alerts.opensearch_index` may be None → skip bulk write if None.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `run_detection_cycle()` below `bulk_write_alerts()`.

**Inputs:**
- `session`: authenticated session.
- `base_url`: cleaned base URL.
- `config`: full resolved config dict from `load_config()`.
- `active_patterns`: `{index_name: [pattern_strings]}` — exclusion patterns per index.
- `args`: parsed CLI args (for `dry_run`).

**Interface Contract:**
```python
def run_detection_cycle(
session: requests.Session,
base_url: str,
config: dict,
active_patterns: dict[str, list[str]],
args: argparse.Namespace
) -> int:
"""
Run full detection cycle across all indexes.
Returns total count of silent identifiers detected (for logging in daemon loop).
Internally builds alert docs and calls bulk_write_alerts().
"""
```

Per-index loop logic:
```
detection_run_id = datetime.datetime.now(utc).strftime('%Y%m%d-%H%M%S')
now = datetime.datetime.now(utc)

for index_cfg in config['indexes']:
primary_results = run_index_pass(session, base_url, index_cfg['name'],
index_cfg['host_field'], index_cfg['baseline_days'])
secondary_results = {}
if index_cfg['secondary_host_field']:
secondary_results = run_index_pass(session, base_url, index_cfg['name'],
index_cfg['secondary_host_field'], index_cfg['baseline_days'])

patterns = active_patterns.get(index_cfg['name'], [])

# Apply exclusions BEFORE consolidation
primary_filtered = {k: v for k, v in primary_results.items()
if not is_excluded(k, patterns)}
secondary_filtered = {k: v for k, v in secondary_results.items()
if not is_excluded(k, patterns)}

silent = consolidate_results(primary_filtered, secondary_filtered,
index_cfg['silence_hours'], index_cfg['name'], now)

for entry in silent:
doc = build_alert_doc(entry, index_cfg, config, detection_run_id,
index_cfg['host_field'], index_cfg['secondary_host_field'], now)
all_alert_docs.append(doc)

bulk_write_alerts(session, base_url,
config['alerts']['opensearch_index'],
all_alert_docs, args.dry_run)
```

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- Exclusions MUST be applied BEFORE calling `consolidate_results()` (on the raw pass results), not after.
- `bulk_write_alerts()` is called ONCE after all indexes are processed, not per-index.
- If `config['alerts']['opensearch_index']` is None, do NOT call `bulk_write_alerts()`.
- `detection_run_id` and `now` MUST be computed once per cycle, not per index.
- Log a summary at INFO level after each index: `"Index logs-*: 12 silent identifiers detected"`.
- Log a summary at INFO level after all indexes: `"Detection cycle complete. Total silent: 42. detection_run_id=20260324-091532"`.

**Must NOT do:**
- Do not reload config here — that is TASK-11's job.
- Do not modify any of the functions from previous tasks.

**Acceptance Criteria:**
- [ ] Secondary pass skipped (no extra query) when `secondary_host_field` is None.
- [ ] Excluded identifiers do NOT appear in alert docs.
- [ ] `bulk_write_alerts` called once with all docs across all indexes.
- [ ] `bulk_write_alerts` NOT called when `alerts.opensearch_index` is None.
- [ ] `detection_run_id` is the same string across all alert docs in one cycle.
- [ ] Return value equals total count of alert docs generated.

**Edge Cases to Handle:**
- All indexes return empty results → `bulk_write_alerts` called with `[]` → function exits early (TASK-09 handles this).
- One index query fails → `run_index_pass()` returns `{}`, cycle continues with other indexes.

**Known Risks / Likely Mistakes:**
- AI may call `bulk_write_alerts` inside the per-index loop → must be outside the loop.
- AI may apply exclusions AFTER consolidation → spec requires exclusions applied to raw pass results before merging.

---

### TASK-11: Daemon Loop

**Objective:** Implement `daemon_loop(args: argparse.Namespace, session: requests.Session, base_url: str) -> None` that handles the run-once vs. long-running daemon logic, config reload, sleep, and cycle logging.

**Bootstrap Context:**
TASK-10 must be complete. Read Section 3 (Runtime Behaviour) of the design spec.
Key facts:
- `sleep_after_run_seconds` source: CLI `args.sleep_seconds` (if not None) overrides `config['sleep_after_run_seconds']`.
- `0` = run once and exit. `> 0` = infinite loop with `time.sleep()`.
- Config is reloaded at the START of every cycle (including the first).
- Expiry filtering via `build_exclusion_set()` is called per cycle.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `daemon_loop()` below `run_detection_cycle()`.

**Interface Contract:**
```python
def daemon_loop(
args: argparse.Namespace,
session: requests.Session,
base_url: str
) -> None:
"""
Infinite loop (or single run) implementing config reload, detection, sleep.
Exits via sys.exit(0) on run-once completion.
On config load failure, logs CRITICAL and exits.
"""
```

Loop logic:
```
while True:
try:
config = load_config(args.config)
except Exception as e:
logging.critical(f"Failed to load config: {e}")
sys.exit(1)

sleep_secs = args.sleep_seconds if args.sleep_seconds is not None else config['sleep_after_run_seconds']

active_patterns = {}
for index_cfg in config['indexes']:
active_patterns[index_cfg['name']] = build_exclusion_set(index_cfg['exclusions'])

run_detection_cycle(session, base_url, config, active_patterns, args)

if sleep_secs == 0:
sys.exit(0)

logging.info(f"Sleeping for {sleep_secs}s before next cycle...")
time.sleep(sleep_secs)
```

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- Config MUST be loaded at the top of EVERY loop iteration (not once before the loop).
- `time.sleep()` is the ONLY permitted sleep mechanism — no threads, no asyncio.
- Config load failure MUST call `sys.exit(1)` — the script must not silently continue with stale config.
- CLI `args.sleep_seconds` takes precedence over `config['sleep_after_run_seconds']` when set.

**Must NOT do:**
- Do not rebuild the `requests.Session` on each cycle — it was built once in TASK-05 and reused.
- Do not modify `run_detection_cycle()`.

**Acceptance Criteria:**
- [ ] `sleep_after_run_seconds: 0` (and `args.sleep_seconds` None) → runs once, exits with code 0.
- [ ] `args.sleep_seconds = 0` (CLI `--sleep 0`) → runs once, exits with code 0, ignoring config sleep value.
- [ ] `sleep_after_run_seconds: 3600` → after first cycle, `time.sleep(3600)` is called, then config is reloaded.
- [ ] Config YAML syntax error on second cycle → logs CRITICAL, `sys.exit(1)`.
- [ ] `args.sleep_seconds = 60` overrides `sleep_after_run_seconds: 86400` from config.

**Known Risks / Likely Mistakes:**
- AI may load config once outside the loop → config must reload every iteration.
- AI may call `sys.exit(0)` after every cycle regardless of sleep setting → only exit if `sleep_secs == 0`.

---

### TASK-12: main() Entry Point & Final Wiring

**Objective:** Implement `main()` that calls `parse_args()`, `setup_logging()`, `build_session()`, and `daemon_loop()`, and add `if __name__ == "__main__": main()` at the bottom of the file. Perform a final review of all imports and ensure the file runs end-to-end.

**Bootstrap Context:**
All previous tasks must be complete. The file now contains all functions. This task wires them together and ensures nothing is missing.

**Files to Create / Modify:**
- `silent_log_detector.py` — MODIFY — Add `main()` and `if __name__ == "__main__"` guard.

**Required imports (complete list — add all at top of file):**
```python
#!/usr/bin/env python3
import argparse
import datetime
import fnmatch
import getpass
import json
import logging
import logging.handlers
import os
import sys
import time
from urllib.parse import quote

import requests
import yaml
```

**Interface Contract:**
```python
def main() -> None:
args = parse_args()
setup_logging(args.verbose)
session, base_url = build_session(args)
daemon_loop(args, session, base_url)
```

⚠️ CRITICAL CONSTRAINTS — THESE MUST NOT BE VIOLATED:
- `setup_logging()` MUST be called before any `logging.*` call.
- `build_session()` MUST be called once and the session reused across all cycles.
- The `if __name__ == "__main__"` guard MUST be present.
- The shebang line `#!/usr/bin/env python3` MUST be the first line of the file.

**Must NOT do:**
- Do not refactor or alter any of the previously implemented functions — this task is wiring only.
- Do not add any new logic to `main()` beyond the four calls listed above.

**Acceptance Criteria:**
- [ ] `python silent_log_detector.py --help` exits 0 and shows all flags.
- [ ] `python silent_log_detector.py` (no args) exits 2 with argparse error about missing `--opensearch-url`.
- [ ] `python silent_log_detector.py --opensearch-url http://localhost:9200 --dry-run --config ./silent-log-detector.yaml` runs without import errors (config file absence produces a clean error, not a traceback).
- [ ] `python silent_log_detector.py --version` prints version string and exits 0.
- [ ] All imports resolve (no `ModuleNotFoundError` for stdlib modules).
- [ ] `PyYAML` and `requests` must be installed: `pip install requests pyyaml`.

**Edge Cases to Handle:**
- If `setup_logging()` raises (e.g. bad log directory), let it propagate — no silent swallow.

**Known Risks / Likely Mistakes:**
- AI may add complex logic to `main()` that belongs in `daemon_loop()` → main() has exactly 4 lines of logic.
- AI may call `load_config()` in `main()` before `daemon_loop()` → config loading belongs inside the loop.

---

## Risk Register

| # | Risk | Severity | Mitigation |
|---|------|----------|-----------|
| R1 | Naive datetime used anywhere → DST/timezone bugs in silence calculation | HIGH | Invariant: all datetimes must be `datetime.datetime.now(datetime.timezone.utc)`. UTCFormatter uses `time.gmtime`. |
| R2 | Exclusions merged instead of replaced at per-index level | MEDIUM | TASK-03 acceptance criteria explicitly test this. |
| R3 | `_doc` POST used instead of `_bulk` → breaks at scale | HIGH | TASK-09 must pass acceptance criteria: no `_doc` POST anywhere in file. |
| R4 | `Content-Type: application/json` on bulk POST → OpenSearch rejects NDJSON | HIGH | Explicitly specified as `application/x-ndjson` in TASK-09. |
| R5 | URL index name not encoded → `*` in `logs-*` breaks HTTP request | MEDIUM | TASK-06 requires `urllib.parse.quote(index_name, safe='')`. |
| R6 | Password logged at DEBUG level | HIGH | TASK-05 explicit constraint: never log password. |
| R7 | `--insecure` silently disables SSL with no warning | MEDIUM | TASK-05 requires `sys.stderr` warning before setting `verify=False`. |
| R8 | Config loaded once before daemon loop, stale after YAML edits | HIGH | TASK-11 loads config inside the loop, every iteration. |
| R9 | `bulk_write_alerts` called once per index instead of once per cycle | MEDIUM | TASK-10 explicit: `bulk_write_alerts` is called outside the per-index loop. |
| R10 | PyYAML not listed as a dependency — user gets `ModuleNotFoundError` | LOW | TASK-12 documents `pip install requests pyyaml`. Surface this assumption to the user. |