nextron_thor_apt_scanner: quality improvements

[navnit-elastic]

Proposed commit message

nextron_thor_apt_scanner: quality improvements for thor_forwarding

CEL improvements
- Rework CEL program to use offset-based pagination with a scan worklist pattern
- Publish events and update cursor per scan instead of waiting for a full search page,
  reducing initial ingestion latency
- Set want_more to false on API error paths and skip already-processed scans via latest_scan_id
- Add proxy_url and ssl stream variables in manifest.yml and wire them into the CEL input template
- Remove unused preserve_duplicate_custom_fields variable

Ingest pipeline improvements
- Bump ECS reference to 9.4.0 in build.yml and ingest pipeline
- Parse thor.end_time with a date processor into event.end instead of rename
- Remove Agentless metadata fields (organization, division, team) from incoming events
- Add on_failure handling to the categorize sub-pipeline
- Add allow_duplicates: false to related.hash and related.user append processors

Field mappings & documentation improvements
- Move group field under thor.files.group and remove redundant ecs.yml declarations
- Add beats.yml and event.module/event.dataset to base-fields.yml
- Clarify batch_size title and description in data stream manifest
- Expand system test mocks for multi-page pagination and update hit_count to 18
- Update pipeline expected outputs, sample_event.json, and generated README

Checklist

• I have reviewed tips for building integrations and this pull request is aligned with them.
• I have verified that all data streams collect metrics or logs.
• I have added an entry to my package's changelog.yml file.
• I have verified that Kibana version constraints are current according to guidelines.
• I have verified that any added dashboard complies with Kibana's Dashboard good practices

Author's Checklist

• Verify that the Agent is able to pick the old cursor with no errors and no events loss on upgrade

How to test this PR locally

Related issues

• Closes nextron_thor_apt_scanner: quality improvements #19901

Screenshots

[github-actions]

✅ Elastic Docs Style Checker (Vale)

No issues found on modified lines!

---

The Vale linter checks documentation changes against the Elastic Docs style guide. To use Vale locally or report issues, refer to Elastic style guide for Vale.

[infra-vault-gh-plugin-prod]

Pinging @elastic/security-service-integrations (Team:Security-Service Integrations)

[vera-review-bot]

Severity: 🟠 High confidence: medium path: packages/nextron_thor/data_stream/thor_forwarding/agent/stream/cel.yml.hbs:179

A scan-search HTTP error is silently swallowed when the worklist was drained in the prior cycle; guard the second stage so the error is surfaced instead of overwritten.

Details

The program runs in two stages against a single, merged state. Stage 1 (scan search) sets worklist/more_pages on success, or on a non-200 returns the single-object error map {"events": {"error": {...}}, "offset": 0, "want_more": false} WITHOUT a worklist key. Stage 2 then runs unconditionally as state.with(<expr>).

Because state.with() preserves keys not present in the merged map, the worklist from the previous cycle survives a Stage 1 error. In normal steady state that surviving value is the empty list [] left by tail(state.worklist, 1) after the last log was processed. So after a search error, Stage 2 evaluates has(state.worklist) = true and size(state.worklist) > 0 = false, and falls into the empty-worklist branch (this line). That branch returns {"events": [], "offset": ..., "want_more": state.more_pages, "cursor": {...}}, which is merged over the state and OVERWRITES events (the error object) with [].

Result: the scan-search API error is never indexed (no error event is emitted), and want_more/offset are driven by the stale state.more_pages from the prior successful search. When that stale more_pages is true, want_more becomes true and offset advances by batch_size on an error cycle, so the program skips past the failed page instead of retrying it. This defeats the error handling that Stage 1 carefully constructs and can cause silent data loss on transient search-API failures.

Recommendation:

Short-circuit the second stage when the first stage already produced an error, so the single-object error map is returned as-is instead of being overwritten by the empty-worklist branch:

  ).as(state,
    // If the scan-search stage already produced an error, surface it and halt;
    // do not let the empty-worklist branch overwrite it with an empty page.
    (has(state.events) && type(state.events) == map) ?
      state
    :
      state.with(
        !has(state.worklist) ?
          state
        : (size(state.worklist) > 0) ?
          request(
            "GET",
            state.url.trim_right("/") + "/v1/scan/log?" + {
              "scan": [state.worklist[0].id],
              "log": ["thor.json"],
            }.format_query()
          )
          // ... unchanged ...
        :
          {
            "events": [],
            "offset": state.more_pages ? (int(state.offset) + int(state.batch_size)) : 0,
            "want_more": state.more_pages,
            "cursor": {
              ?"latest_scan_ts": state.?cursor.latest_scan_ts,
              ?"latest_scan_id": state.?cursor.latest_scan_id,
            },
          }
      )
  )

(On a successful search the first stage sets worklist/more_pages and does not set events, so the guard only triggers on the error map.)

---

🤖 AI-Generated Review | Vera Review Bot | 📚 Knowledge base: integration-skills

⚠️ Automated review — verify suggestions before applying.

[vera-review-bot]

Severity: 🟠 High confidence: medium path: packages/nextron_thor/data_stream/thor_forwarding/agent/stream/cel.yml.hbs:150

The scan-search offset is advanced but never reset when a collection cycle ends on a partial last page, so the next periodic run skips scans; only advance offset when more_pages is true, else reset it to 0.

Details

When the worklist drains its last item (size(state.worklist) == 1) this branch sets offset = int(state.offset) + int(state.batch_size) unconditionally, and sets want_more = size(state.worklist) > 1 || state.more_pages. On the final, partial page of a cycle more_pages is false (body.data.size() < batch_size), so evaluation terminates here with want_more=false AND a non-zero offset. offset is a top-level (non-cursor) state key, so it is retained across periodic evaluations in the running input session. The offset is only ever reset to 0 in the empty-worklist fallthrough branch and the error branches, neither of which is reached on this path. Consequently the next periodic run re-initializes start_time from the cursor but issues /v1/scan/search with the stale offset, causing the server to skip the first offset matching scans of the new time window — those are new, unprocessed scans (the already-processed one is dropped separately by the s.id != latest_scan_id client filter), so they are silently lost. This only self-corrects when the final page is exactly full (total a multiple of batch_size), because then an extra empty-page query routes through the reset branch. The typical partial-final-page case loses data every cycle after the first.

Recommendation:

Only advance offset when there is genuinely a next page (more_pages), and reset it to 0 when the current page is the last one, so a fresh periodic cycle starts at offset 0:

"offset": (size(state.worklist) == 1) ?
  (state.more_pages ? (int(state.offset) + int(state.batch_size)) : 0)
:
  state.offset,

---

🤖 AI-Generated Review | Vera Review Bot | 📚 Knowledge base: integration-skills

⚠️ Automated review — verify suggestions before applying.

[navnit-elastic]

Good catch. offset is retained across periodic evaluations within a session. Line 150 unconditionally advances offset when draining the last worklist item, so a partial final page can leave a non-zero offset into the next interval.

[elastic-vault-github-plugin-prod]

🚀 Benchmarks report

To see the full report comment with /test benchmark fullreport

[vera-review-bot]

No issues across the latest commits dbbf919.

Review summary

Issues found across earlier commits [b4b9d68](https://github.com/elastic/integrations/commit/b4b9d685b8e343896c34b8710622af04fc5e4738) — 1 high

• 🟠 The scan-search offset is advanced but never reset when a collection cycle ends on a partial last page, so the next periodic run skips scans; only advance offset when more_pages is true, else reset it to 0. (link) (Unresolved)

Issues found across earlier commits [b758bdd](https://github.com/elastic/integrations/commit/b758bdd45158edede2dc507fb419fe3e7d326ec3) — 1 high

• 🟠 A scan-search HTTP error is silently swallowed when the worklist was drained in the prior cycle; guard the second stage so the error is surfaced instead of overwritten. (link) (Unresolved)

A new commit triggers another review — at most once every 15 minutes. I skip the PR while it's approved or has merge conflicts.

🤖 AI-Generated Review | Vera Review Bot | 📚 Knowledge base: integration-skills

⚠️ Automated review — verify suggestions before applying.

[elastic-vault-github-plugin-prod]

✅ All changelog entries have the correct PR link.

[infra-vault-gh-plugin-prod]

💚 Build Succeeded

• Buildkite Build
• Commit: dbbf919

History

• 💚 Build #45444 succeeded 6a3b847
• 💚 Build #45397 succeeded b4b9d68
• 💚 Build #45393 succeeded b758bdd

cc @navnit-elastic