From 420a66fc08be5c5c7ad22f30faaee614c9b0f08c Mon Sep 17 00:00:00 2001 From: Rimuksh Kansal Date: Thu, 10 Sep 2026 21:16:31 +0900 Subject: [PATCH 1/3] add connector-review skill --- .claude/skills/connector-review/SKILL.md | 134 +++++++++++++++++++++++ AGENTS.md | 1 + 2 files changed, 135 insertions(+) create mode 100644 .claude/skills/connector-review/SKILL.md diff --git a/.claude/skills/connector-review/SKILL.md b/.claude/skills/connector-review/SKILL.md new file mode 100644 index 0000000000..1e4bbcc7d5 --- /dev/null +++ b/.claude/skills/connector-review/SKILL.md @@ -0,0 +1,134 @@ +--- +name: connector-review +description: Adversarial 4-expert review of a connectors PR, branch, or ref range (sinks, sources, runtime, SDK, transforms) with clean-room validation of every finding. Experts work alone, no peer debate. Expensive, one run spawns ~10 subagents. Load connectors-overview first. +argument-hint: "[PR number | branch | ref range]" +disable-model-invocation: true +--- + +# Apache Iggy Connector Review + +`` = `$ARGUMENTS`: a PR number, a branch, or a ref range. Empty means `origin/master..HEAD`. Scope: anything under `core/connectors/` plus connector integration tests under `core/integration/tests/connectors/`. + +Prereq: load `connectors-overview` first as router (sibling skill in this directory; repo-wide rules in `/AGENTS.md`). This file owns **review mechanics only**. + +You = **moderator**. You never open the diff or a source file: you route paths, merge claims, synthesize. Every token you load rides along every later turn. Reviewers and validators are one-shot agents that deliver by writing a file; nobody chats. + +## Charter (paste VERBATIM into every expert, validator, and tiebreak prompt) + +> You think big brain. You speak caveman. Separate things. +> +> **Thinking, unchanged.** Read the diff, then every changed file in full from the local checkout, then whatever call sites you need. Trace call chains. Verify invariants. Prove findings, don't guess. Cite `path::Symbol`, plus `file:line` as secondary anchor. Running tests or builds needs a stated justification: reading and tracing settles most claims, and parallel cargo runs block on one target-dir lock. +> +> **Output style.** Drop articles, filler, pleasantries, hedging. Fragments OK. Keep EXACT: symbol, `file:line`, error quotes, code, technical terms, severity and confidence labels. +> +> - Finding, one line each: `[sev] path::Symbol (file:line) - problem. Fix: action. (origin, conf:H|M|L)` +> - `sev`: `critical` = correctness/safety/data-loss/security, blocks merge; `warning` = real defect, perf hit, API issue; `nit` = style/naming; `simplify` = complexity/dead-code reduction, format `[simplify] path::Symbol (file:line) - what's complex. Simpler: alternative. Saves: ~N lines / removes indirection. (origin, conf)`. +> - `origin`: `intro` (PR introduced), `pre-surfaced` (existed, exposed by PR), `pre-untouched` (existed, not touched). +> - Never flag em dashes or other punctuation style as a finding. +> - Simplification mandate: less code > more code. Per changed file ask whether ~30% smaller keeps correctness: dead fields/params/branches/imports, duplication of an existing helper (cite it), single-impl traits, premature generics, checks for impossible states. Do not propose simplifications that change semantics or break public API. If nothing qualifies, write `Simplifications: none`. +> +> **Connector mandatory checks (every expert, where in scope).** Verify, don't assume: +> `SecretString` on credential fields, no `Serialize` on plugin config (or redacted serializer with justification); FFI return codes honored (`0` ok, nonzero fail; duplicate-ID guard intact); sink redelivery claim vs `AutoCommit::When(PollingMessages)` + FFI swallow path; idempotency key on retry (message `id` dedup, `ON CONFLICT`, composite `_id`, deterministic run id + 409 handling). +> `Permanent*` vs transient classification vs retry strategy agreement; drop accounting (`errors` vs `messages_filtered`, flushed once per batch via pre-built labels). +> Config forward-compat (`Option`/`#[serde(default)]`); `consume`/`poll` take `&self`; no `tokio::spawn`/`block_on` in plugins; no lock held across `.await`; `BTreeMap` headers; `try_to_bytes(&self)` no-clone for JSON; `Vec::with_capacity`; `[lib] crate-type = ["cdylib", "lib"]`; benchmark/verbose flags wired; closest exemplar named. +> +> Caveman = output compression, not analysis compression. Dig deep. Write short. + +## Step 1: Identify the target (no reading) + +- Classify ``: matches `^#?(pr)?[0-9]+$` case-insensitively -> PR, the digits are ``. Anything else -> ref range or bare branch. Empty -> review `origin/master..HEAD`. +- ``: `` lowercased, chars outside `[a-z0-9-]` replaced by `-`, repeats collapsed, trimmed, max 40 chars (`PR3123` -> `pr3123`, `origin/master..HEAD` -> `origin-master-head`). Empty -> `date +%s`. +- `` = `/review-`. `mkdir -p` it. +- PR: `gh pr view --json title,body,headRefOid > /pr.json`, `gh pr diff > /diff.patch`, `gh pr diff --name-only > /files.txt`. `` = first 8 of `headRefOid`. +- Ref range or bare branch: `git diff $(git merge-base origin/master HEAD)..HEAD > /diff.patch`, same with `--name-only`, `` = `git rev-parse --short=8 HEAD`. No `pr.json` on this path. +- Guard: `git rev-parse HEAD` must equal the reviewed head. Experts read the local checkout; if it differs, stop and ask the user to check out the reviewed head. +- ``: 1-3 word `snake_case` summary, `[a-z0-9_]`, <= 24 chars. From the PR title; no PR -> from `git log -1 --format=%s`. +- Report path: `/report.md`. + +Do not `cat` any of the files you just wrote. `wc -l /diff.patch` is the only look you take. + +## Step 2: Round 1, four one-shot experts (one message, parallel) + +Spawn 4 `Agent` calls in a single message: `subagent_type: general-purpose`, `name: -` (bare role names collide with concurrent sessions: one shared agent namespace), no `model` (inherits). Prompt = role block + Charter + this brief, with ``, ``, `` filled in: + +> Target: `` at ``. Diff: `/diff.patch`. Changed files: `/files.txt`. PR title and body: `/pr.json` (drop this sentence when there is no PR). Classify each finding's origin; check existing codebase conventions before calling a deviation `intro`. Name the closest exemplar plugin compared (`stdout_sink`, `postgres_sink`, `http_sink`, `random_source`, ...). +> Deliverable = the file `/.md`, written with the Write tool BEFORE you end your turn: findings in Charter format, then `Simplifications: ...`, then `Verdict: APPROVE | REQUEST CHANGES - reason`. A previous worker finished reading and then idled without delivering; the Write call IS the delivery, your final message is just the path. Budget 3/4 reading, 1/4 writing; partial beats unshipped. +> You work alone: no teammates, no SendMessage, no questions back. + +Role blocks: + +- **plugin**: Senior connector-plugin engineer. Owns sink/source impls. + Focus: lifecycle (`&self` consume/poll, `open` connectivity fail-fast, `close` flush/take), config `Option` defaults + `new` vs `open` vs `consume` validation order, payload dispatch + `try_to_bytes` + `mem::replace` + `with_capacity`, header `BTreeMap` handling, error-variant mapping (transient vs `Permanent*`), retry strategy agreement, idempotency on redelivery, batching (`chunks`, no cross-`consume` buffering), logging redaction + `verbose_logging`. + Simplify: dead payload arms, skip branches that never fire, single-variant enums, near-duplicate path/config helpers. +- **runtime**: Connectors runtime + FFI host engineer. Owns `runtime/src/`. + Focus: FFI pointer lifetimes, `plugin_id` identity, container `Arc` ownership (no unload mid-call), sink consume loop (autocommit timing, decode/transform drop-and-continue, postcard encode, nonzero-return handling), source forwarding loop (flume handoff, `cleanup_sender` order, state save after Iggy send), state atomic-rename protocol, `restart_guard`, metrics/drop-accounting wired sites, benchmark event + histogram labels, config provider + `ConfigEnv` addressing. + Simplify: duplicated consume/forward paths, label lookups per message, re-expanded log-layer matches. +- **sdk**: SDK contract guardian. Owns `sdk/src/`. + Focus: `#[repr(C)]` layouts, `Schema`/`Payload`/`Error` variant changes, decoder/encoder round-trips, `sink_connector!`/`source_connector!` return codes + duplicate-ID guard, `ConnectorState` msgpack format, `retry.rs` (breaker, middleware, `max_retries` total-vs-extra semantics), `convert.rs` bridges, transform trait (`&self`, sync, `Ok(None)` filter vs `Err`), ser/de split (postcard FFI, JSON config, msgpack state). + Simplify: single-impl traits, premature generics, predicates enforced twice. +- **testing**: Connector test + docs lead. + Focus: BDD unit naming consistency per file, `test_config()` helper, four canonical source state tests, sink pure-logic coverage (defaults, payload variants, header encoding, query building, transient/permanent classification), transform four branches, integration layout (`#[iggy_harness]` + `testcontainers-modules` + `iggy-test-` prefix + polling not sleeping). + Also: README + `example_config` TOML sync (all fields incl. `max_connections`, retry knobs, benchmark/verbose), error-message clarity, breaking-change notes. + Simplify: mocked-backend tests that should be real-infra, sleeps that should poll, `Display`-string assertions. + +Collect: wait for the completion notifications, then `ls /*.md`. +A role with no file gets one `SendMessage` nudge to `-` ("Write `/.md` now, then stop."); still missing after that, respawn the role once with the same prompt. Never open a subagent transcript via `TaskOutput` (it is the whole JSONL). + +## Step 3: Merge into neutral claims (moderator) + +Read the 4 role files. Write `/claims.md`, one line per claim: `C [sev] path::Symbol (file:line) - claim. Fix: action. (origin)`. Strip role names, confidence, and argument. Same symbol + same defect from several roles = one claim at the highest severity; keep a private raised-by map for the report. Simplify items are claims too. + +No claims at all: skip Steps 4 and 5, go to Step 6 with empty sections and `Verdict: APPROVE`. The report file still gets written. + +## Step 4: Clean-room validation (one message, parallel) + +Shard claims ~5 per validator. Spawn one `Agent` per shard plus one sweep validator, all in one message: `subagent_type: general-purpose`, `model: opus`, `name: validator--` / `sweep-`. Each gets ONLY: its claims verbatim, `/files.txt`, `/diff.patch`, the target identity, the Charter. Not the role files, not raised-by, not your reasoning; the missing context is what removes the anchoring bias. + +Validator mandate (adversarial): for each claim open the cited symbol, trace call sites, then rate `C: PASS | FIX: | REMOVE: `; judge whether the severity is calibrated; re-check the anchor. Deliverable `/validate-.md` via Write, same idle rule as Step 2. + +Sweep mandate: all claims + the diff. Two questions only: which real defects in the diff are missing from the list, and which listed items wrongly clear a bug. Deliverable `/sweep.md`, additions in Charter format tagged `(sweep)`. + +Apply: drop REMOVE, apply FIX (wording, symbol, line, severity), fold sweep additions in as `(sweep, unvalidated)`. A `critical` sweep addition gets one extra validator before it may block the verdict. + +## Step 5: Contested items (only when triggered) + +Contested = a validator REMOVEs or downgrades a `critical` or `warning`, or a sweep addition contradicts a PASS. Per item spawn one `Agent` (`model: opus`) with the claim, the validator's verdict text, the expert's original line, and the paths; it writes `UPHELD | OVERTURNED - reason (cite symbol)` to `/contested-.md`. Cap 5 per run; past the cap you adjudicate and mark `(moderator call)`. + +## Step 6: Synthesize, write, done + +Output in caveman style: + +```text +## Review: [change desc] + +### Confirmed (expert + clean-room validator) +- [sev] path::Symbol (file:line) - problem. Fix: action. (raised: role[, role]; validated: PASS|FIX) + +### Contested +- path::Symbol (file:line) - problem. + Expert: position. Validator: counter. **Tiebreak**: UPHELD|OVERTURNED - why. + +### Retracted (validator REMOVE) +- finding - why. + +### Pre-existing (origin pre-*, not blocking) +- path::Symbol (file:line) - follows pattern in [ref]. + +### Simplification opportunities (non-blocking) +- path::Symbol (file:line) - current shape. Simpler: alternative. Saves: ~N lines / removes indirection. + +### Verdict: APPROVE | REQUEST CHANGES +Confirmed critical + warning only. Simplifications informational. Reason: one line. + +Counts: critical N, warning N, nit N, simplify N (Confirmed + Simplification sections) +``` + +Then write `/report.md` with: + +1. H1 `# Connector Review - ()`. +2. Metadata, one line each: target ``, reviewed commit, ISO timestamp, roles, validator count, contested count. +3. The report above, verbatim. +4. Appendix `## Raw findings per expert`: each role file verbatim in a fenced block. +5. `## Validation record`: counts of PASS / FIX / REMOVE, sweep additions, contested outcomes. + +Last user-facing line: `Findings written: /report.md`. No cleanup: one-shot agents end themselves, `` stays in the scratchpad. diff --git a/AGENTS.md b/AGENTS.md index ada37f1457..2eea76aebf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -130,6 +130,7 @@ connectors work, load `connectors-overview` first as router. - [connector-source](.claude/skills/connector-source/SKILL.md) - source plugin authoring - [connector-transform](.claude/skills/connector-transform/SKILL.md) - transform authoring - [connector-testing](.claude/skills/connector-testing/SKILL.md) - unit + integration test patterns +- [connector-review](.claude/skills/connector-review/SKILL.md) - adversarial 4-expert connectors PR/branch review, ~10 subagents per run Repo-wide, user-invoked only. `disable-model-invocation: true` keeps it out of the agent's context; do not replicate its steps. When a From d5b1d7f099ed4c95543b5b53e81a56be449145dc Mon Sep 17 00:00:00 2001 From: Rimuksh Kansal Date: Thu, 10 Sep 2026 21:33:26 +0900 Subject: [PATCH 2/3] add more details to skill --- .claude/skills/connector-review/SKILL.md | 35 ++++++++++++++++++------ 1 file changed, 26 insertions(+), 9 deletions(-) diff --git a/.claude/skills/connector-review/SKILL.md b/.claude/skills/connector-review/SKILL.md index 1e4bbcc7d5..f95912a216 100644 --- a/.claude/skills/connector-review/SKILL.md +++ b/.claude/skills/connector-review/SKILL.md @@ -28,9 +28,12 @@ You = **moderator**. You never open the diff or a source file: you route paths, > - Simplification mandate: less code > more code. Per changed file ask whether ~30% smaller keeps correctness: dead fields/params/branches/imports, duplication of an existing helper (cite it), single-impl traits, premature generics, checks for impossible states. Do not propose simplifications that change semantics or break public API. If nothing qualifies, write `Simplifications: none`. > > **Connector mandatory checks (every expert, where in scope).** Verify, don't assume: -> `SecretString` on credential fields, no `Serialize` on plugin config (or redacted serializer with justification); FFI return codes honored (`0` ok, nonzero fail; duplicate-ID guard intact); sink redelivery claim vs `AutoCommit::When(PollingMessages)` + FFI swallow path; idempotency key on retry (message `id` dedup, `ON CONFLICT`, composite `_id`, deterministic run id + 409 handling). -> `Permanent*` vs transient classification vs retry strategy agreement; drop accounting (`errors` vs `messages_filtered`, flushed once per batch via pre-built labels). -> Config forward-compat (`Option`/`#[serde(default)]`); `consume`/`poll` take `&self`; no `tokio::spawn`/`block_on` in plugins; no lock held across `.await`; `BTreeMap` headers; `try_to_bytes(&self)` no-clone for JSON; `Vec::with_capacity`; `[lib] crate-type = ["cdylib", "lib"]`; benchmark/verbose flags wired; closest exemplar named. +> `SecretString` on credential fields, no `Serialize` on plugin config (or redacted serializer with justification); FFI return codes honored (`0` ok, `-1` invalid, `1` open failure; duplicate-ID guard intact); sink redelivery claim vs `AutoCommit::When(PollingMessages)` + FFI swallow path; idempotency key on retry (message `id` dedup, `ON CONFLICT`, composite `_id`, deterministic run id + 409 handling). +> `Permanent*` vs transient classification vs retry strategy agreement; drop accounting (`errors` vs `messages_filtered`, flushed once per batch via pre-built labels); every error lands at its level (fatal exits process, per-plugin sets status, per-message skips, per-batch bumps metric or logs). +> Config forward-compat (`Option`/`#[serde(default)]`); `consume`/`poll` take `&self`; no `tokio::spawn`/`block_on` in plugins; no lock held across `.await`; `BTreeMap` headers; `try_to_bytes(&self)` no-clone for JSON; `simd_json::OwnedValue` mutated in place, never clone-then-replace; `Vec::with_capacity`; `[lib] crate-type = ["cdylib", "lib"]`; benchmark/verbose flags wired; closest exemplar named. +> STOP tripwires are critical: `#[repr(C)]` change without SDK bump; `Schema` variant rename; transient↔`Permanent*` promotion; consumer-group rename; plugin path resolution change; trait change without changelog + all-plugins-same-PR + bump + old-`.so` note. +> `state.rs` protocol change is critical: 0o600 mode, tmp `sync_data`, post-rename parent-dir `sync_all`, save/load failures surface (`CannotWriteStateFile`/`CannotOpenStateFile`), never silent fresh-start. Scope creep (unrelated refactor in same PR) gets flagged. +> Logging shape: connector ID + name on every line, literal API field labels, `debug!` default upgraded to `info!` on verbose only, never eager `format!()` around tracing args, never log secrets. Control API returns plugin config verbatim (#3802): never demand plugin-side fixes for control-plane exposure. > > Caveman = output compression, not analysis compression. Dig deep. Write short. @@ -51,25 +54,39 @@ Do not `cat` any of the files you just wrote. `wc -l /diff.patch` is the on Spawn 4 `Agent` calls in a single message: `subagent_type: general-purpose`, `name: -` (bare role names collide with concurrent sessions: one shared agent namespace), no `model` (inherits). Prompt = role block + Charter + this brief, with ``, ``, `` filled in: -> Target: `` at ``. Diff: `/diff.patch`. Changed files: `/files.txt`. PR title and body: `/pr.json` (drop this sentence when there is no PR). Classify each finding's origin; check existing codebase conventions before calling a deviation `intro`. Name the closest exemplar plugin compared (`stdout_sink`, `postgres_sink`, `http_sink`, `random_source`, ...). +> Target: `` at ``. Diff: `/diff.patch`. Changed files: `/files.txt`. PR title and body: `/pr.json` (drop this sentence when there is no PR). Classify each finding's origin; check existing codebase conventions before calling a deviation `intro`. Name the closest exemplar plugin compared (`stdout_sink`, `postgres_sink`/`postgres_source`, `http_sink`, `mongodb_sink`, `random_source`, ...). > Deliverable = the file `/.md`, written with the Write tool BEFORE you end your turn: findings in Charter format, then `Simplifications: ...`, then `Verdict: APPROVE | REQUEST CHANGES - reason`. A previous worker finished reading and then idled without delivering; the Write call IS the delivery, your final message is just the path. Budget 3/4 reading, 1/4 writing; partial beats unshipped. > You work alone: no teammates, no SendMessage, no questions back. Role blocks: - **plugin**: Senior connector-plugin engineer. Owns sink/source impls. - Focus: lifecycle (`&self` consume/poll, `open` connectivity fail-fast, `close` flush/take), config `Option` defaults + `new` vs `open` vs `consume` validation order, payload dispatch + `try_to_bytes` + `mem::replace` + `with_capacity`, header `BTreeMap` handling, error-variant mapping (transient vs `Permanent*`), retry strategy agreement, idempotency on redelivery, batching (`chunks`, no cross-`consume` buffering), logging redaction + `verbose_logging`. + Sink focus: lifecycle (`open` connectivity fail-fast `InitError`, `close` flush + `.take()` + final stats), config `Option` defaults + conflict fix in `new()` + `warn!` (never error, never validate in `consume()`), durations `Option` humantime (no `humantime_serde`). + Sink payloads/errors: dispatch Json/Raw/Text minimum (`InvalidPayloadType` or base64 fallback), `try_to_bytes` + `mem::replace` + `with_capacity`, header `is_empty` check + binary-base64 vs text split in `BTreeMap`, `InvalidRecordValue` skip+log, `WriteFailure` caller-decides, `CatalogCommitError` txn consumed and not idempotent. + Sink retry/idempotency: 3-attempt cap, `Retry-After` honored, per-status custom strategy, SQLSTATE-style transient map; dedup-on-write via message `id` (mongo composite `_id` reference; ES auto-`_id` is NOT idempotent); `last_err` pattern (process all batches, return last); batch `chunks` default 100, no cross-`consume` buffering, no config mutation in `consume()`. + Source focus: `poll()` `Err` only logs, loop continues, status never flips (runtime sets `Error` on transform/encode/send/save failure only); always return state incl. empty polls; poll-return-to-save crash window means at-least-once; sleep-first or idle spins CPU; single `poll()` task. + Source state/ids: `ProducedMessage.id` natural ID + `origin_timestamp` nanos, `timestamp`/`checksum` left `None`; brief-lock cursor pattern; state helpers `Option` + log, non-fatal; `State` small and bounded; `SOURCE_SENDERS` cleaned on close; sync poll blocks a worker, `handle` with no timeout stalls one. + Transform focus: three `mod.rs` edits (module, `TransformType`, `from_config`) + `snake_case` TOML match; `InvalidConfigValue` with context; JSON-only with non-JSON passthrough (format converters exempt); compile regex/paths once in `new()`, plain fields, no locks; sync stays sync; stateless `Send + Sync`. + Transform errors: missing field prefers passthrough over `Err` unless config says fail; `Ok(None)` filter logs at `debug!`; no `SystemTime::now()`. + Logging: redaction at construction, `verbose_logging` mirror. Simplify: dead payload arms, skip branches that never fire, single-variant enums, near-duplicate path/config helpers. - **runtime**: Connectors runtime + FFI host engineer. Owns `runtime/src/`. - Focus: FFI pointer lifetimes, `plugin_id` identity, container `Arc` ownership (no unload mid-call), sink consume loop (autocommit timing, decode/transform drop-and-continue, postcard encode, nonzero-return handling), source forwarding loop (flume handoff, `cleanup_sender` order, state save after Iggy send), state atomic-rename protocol, `restart_guard`, metrics/drop-accounting wired sites, benchmark event + histogram labels, config provider + `ConfigEnv` addressing. + Focus: FFI pointer lifetimes (call-duration only), `plugin_id` monotonic never reused, `LogCallback` static, container `Arc` outlives tasks (no unload mid-call), `DashMap` keyed by TOML key + status counters on transitions. + Loops: sink (autocommit timing, `consumer.next()` Err auto-committed drop, decode/transform drop-and-continue, postcard encode, nonzero-return, flush on offset gap, failed batch still emits); source (flume handoff, close-then-`cleanup_sender`-then-drain, state save after Iggy send, transform `Err` never flips status); state atomic-rename protocol; `restart_guard.try_lock()`. + Config/metrics: local discovery + version-wins + `.active_versions`, HTTP 5xx-only retry, `ConfigEnv` skip/leaf; `Counter` names never end `_total`, snake_case label enums, histogram constructors; histograms always on, flag gates text event only. + Hard rules: never block executor, no new static mutables beyond `LOG_CALLBACK`/`PLUGIN_ID`/`SOURCE_SENDERS`, no timeout on FFI without kill-task strategy, no `DashMap` guard across `.await`, duplicate open ID returns `-1`. Simplify: duplicated consume/forward paths, label lookups per message, re-expanded log-layer matches. - **sdk**: SDK contract guardian. Owns `sdk/src/`. - Focus: `#[repr(C)]` layouts, `Schema`/`Payload`/`Error` variant changes, decoder/encoder round-trips, `sink_connector!`/`source_connector!` return codes + duplicate-ID guard, `ConnectorState` msgpack format, `retry.rs` (breaker, middleware, `max_retries` total-vs-extra semantics), `convert.rs` bridges, transform trait (`&self`, sync, `Ok(None)` filter vs `Err`), ser/de split (postcard FFI, JSON config, msgpack state). + Focus: `Send + Sync` on public traits; `#[repr(C)]` layouts (field add needs SDK bump); `Schema` change checklist (enum + `Payload` variant + `try_into_vec`/`try_to_bytes` no-clone + `Display` + `try_into_payload` + factories + modules + README lists + round-trip tests). + Errors/macros: new `Error` only for distinct handling (`String` context, retry docstring, existing variants first); macros return `0`/`-1`/`1`, duplicate-ID guard, `INSTANCES` only global, `version()` static, signature change updates `main.rs` lockstep; stateful decoders need lenient `new` + strict `try_new` + `Default`. + State/retry: `ConnectorState` msgpack opaque, helpers non-fatal, tolerate fresh state; `retry.rs` (`CircuitBreaker.try_lock`, 429/5xx + `Retry-After`, `max_retries` = total attempts, `Duration` API); `convert.rs` bridge; transform trait (`&self`, sync, `Ok(None)` filter vs `Err`); ser/de split (postcard FFI, JSON config, msgpack state). Simplify: single-impl traits, premature generics, predicates enforced twice. - **testing**: Connector test + docs lead. - Focus: BDD unit naming consistency per file, `test_config()` helper, four canonical source state tests, sink pure-logic coverage (defaults, payload variants, header encoding, query building, transient/permanent classification), transform four branches, integration layout (`#[iggy_harness]` + `testcontainers-modules` + `iggy-test-` prefix + polling not sleeping). + Focus: unit BDD `given_when_should` consistency per file + `test_config()` helper + async via local `Runtime::new()` (never `#[tokio::test]` in `src/lib.rs`); four canonical source state tests; sink pure-logic coverage (defaults, payload variants, header encoding, query building, transient/permanent classification); transform four branches. + Integration: declarative `__` naming (never `given_should` there); `#[iggy_harness]` + `TestFixture` + `testcontainers-modules` + `iggy-test-` prefix + polling not sleeping; env injection via `ConfigEnv` forms; `restart.rs` state-survival for stateful sources; cross-boundary tests live in `sdk/tests`, `runtime/`, `api/`. + Discipline: pure logic and config validation stay unit; never mock client lib/server/runtime (`wiremock` exception); `expect()` with step context over bare `unwrap()`; no shared-global coupling; no `Display`-string assertions. Also: README + `example_config` TOML sync (all fields incl. `max_connections`, retry knobs, benchmark/verbose), error-message clarity, breaking-change notes. - Simplify: mocked-backend tests that should be real-infra, sleeps that should poll, `Display`-string assertions. + Simplify: mocked-backend tests that should be real-infra, sleeps that should poll. Collect: wait for the completion notifications, then `ls /*.md`. A role with no file gets one `SendMessage` nudge to `-` ("Write `/.md` now, then stop."); still missing after that, respawn the role once with the same prompt. Never open a subagent transcript via `TaskOutput` (it is the whole JSONL). From 385dad3b6326ae49065ec62575ad3fa73cdb39f2 Mon Sep 17 00:00:00 2001 From: Rimuksh Kansal Date: Tue, 15 Sep 2026 01:46:14 +0900 Subject: [PATCH 3/3] fixes --- .claude/skills/connector-review/SKILL.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.claude/skills/connector-review/SKILL.md b/.claude/skills/connector-review/SKILL.md index f95912a216..b56d1711f6 100644 --- a/.claude/skills/connector-review/SKILL.md +++ b/.claude/skills/connector-review/SKILL.md @@ -43,8 +43,8 @@ You = **moderator**. You never open the diff or a source file: you route paths, - ``: `` lowercased, chars outside `[a-z0-9-]` replaced by `-`, repeats collapsed, trimmed, max 40 chars (`PR3123` -> `pr3123`, `origin/master..HEAD` -> `origin-master-head`). Empty -> `date +%s`. - `` = `/review-`. `mkdir -p` it. - PR: `gh pr view --json title,body,headRefOid > /pr.json`, `gh pr diff > /diff.patch`, `gh pr diff --name-only > /files.txt`. `` = first 8 of `headRefOid`. -- Ref range or bare branch: `git diff $(git merge-base origin/master HEAD)..HEAD > /diff.patch`, same with `--name-only`, `` = `git rev-parse --short=8 HEAD`. No `pr.json` on this path. -- Guard: `git rev-parse HEAD` must equal the reviewed head. Experts read the local checkout; if it differs, stop and ask the user to check out the reviewed head. +- Ref range (`A..B` / `A...B`): `git diff > /diff.patch`, same with `--name-only`, `` = `git rev-parse --short=8 `. Bare branch ``: `git diff $(git merge-base origin/master ).. > /diff.patch`, same with `--name-only`, `` = `git rev-parse --short=8 `. No `pr.json` on either path. +- Guard: `git rev-parse HEAD` must equal the reviewed head (`headRefOid` for a PR, `` for a range, `` for a branch). Experts read the local checkout; if it differs, stop and ask the user to check out the reviewed head. - ``: 1-3 word `snake_case` summary, `[a-z0-9_]`, <= 24 chars. From the PR title; no PR -> from `git log -1 --format=%s`. - Report path: `/report.md`. @@ -64,7 +64,7 @@ Role blocks: Sink focus: lifecycle (`open` connectivity fail-fast `InitError`, `close` flush + `.take()` + final stats), config `Option` defaults + conflict fix in `new()` + `warn!` (never error, never validate in `consume()`), durations `Option` humantime (no `humantime_serde`). Sink payloads/errors: dispatch Json/Raw/Text minimum (`InvalidPayloadType` or base64 fallback), `try_to_bytes` + `mem::replace` + `with_capacity`, header `is_empty` check + binary-base64 vs text split in `BTreeMap`, `InvalidRecordValue` skip+log, `WriteFailure` caller-decides, `CatalogCommitError` txn consumed and not idempotent. Sink retry/idempotency: 3-attempt cap, `Retry-After` honored, per-status custom strategy, SQLSTATE-style transient map; dedup-on-write via message `id` (mongo composite `_id` reference; ES auto-`_id` is NOT idempotent); `last_err` pattern (process all batches, return last); batch `chunks` default 100, no cross-`consume` buffering, no config mutation in `consume()`. - Source focus: `poll()` `Err` only logs, loop continues, status never flips (runtime sets `Error` on transform/encode/send/save failure only); always return state incl. empty polls; poll-return-to-save crash window means at-least-once; sleep-first or idle spins CPU; single `poll()` task. + Source focus: `poll()` `Err` only logs, loop continues, status never flips (runtime sets `Error` on decode/transform/encode/send/save failure only, via `set_error` + `Nack`); always return state incl. empty polls; cursor changes and destructive ops staged in `poll()`, applied only in `Source::on_batch_result` on `Ack`, discarded on `Nack`; sleep-first or idle spins CPU; single `poll()` task. Source state/ids: `ProducedMessage.id` natural ID + `origin_timestamp` nanos, `timestamp`/`checksum` left `None`; brief-lock cursor pattern; state helpers `Option` + log, non-fatal; `State` small and bounded; `SOURCE_SENDERS` cleaned on close; sync poll blocks a worker, `handle` with no timeout stalls one. Transform focus: three `mod.rs` edits (module, `TransformType`, `from_config`) + `snake_case` TOML match; `InvalidConfigValue` with context; JSON-only with non-JSON passthrough (format converters exempt); compile regex/paths once in `new()`, plain fields, no locks; sync stays sync; stateless `Send + Sync`. Transform errors: missing field prefers passthrough over `Err` unless config says fail; `Ok(None)` filter logs at `debug!`; no `SystemTime::now()`. @@ -72,7 +72,7 @@ Role blocks: Simplify: dead payload arms, skip branches that never fire, single-variant enums, near-duplicate path/config helpers. - **runtime**: Connectors runtime + FFI host engineer. Owns `runtime/src/`. Focus: FFI pointer lifetimes (call-duration only), `plugin_id` monotonic never reused, `LogCallback` static, container `Arc` outlives tasks (no unload mid-call), `DashMap` keyed by TOML key + status counters on transitions. - Loops: sink (autocommit timing, `consumer.next()` Err auto-committed drop, decode/transform drop-and-continue, postcard encode, nonzero-return, flush on offset gap, failed batch still emits); source (flume handoff, close-then-`cleanup_sender`-then-drain, state save after Iggy send, transform `Err` never flips status); state atomic-rename protocol; `restart_guard.try_lock()`. + Loops: sink (autocommit timing, `consumer.next()` Err auto-committed drop, decode/transform drop-and-continue, postcard encode, nonzero-return, flush on offset gap, failed batch still emits); source (flume handoff, close-then-`cleanup_sender`-then-drain, state save after Iggy send, decode/transform/encode failure rejects batch via `set_error` + `Nack`); state atomic-rename protocol; `restart_guard.try_lock()`. Config/metrics: local discovery + version-wins + `.active_versions`, HTTP 5xx-only retry, `ConfigEnv` skip/leaf; `Counter` names never end `_total`, snake_case label enums, histogram constructors; histograms always on, flag gates text event only. Hard rules: never block executor, no new static mutables beyond `LOG_CALLBACK`/`PLUGIN_ID`/`SOURCE_SENDERS`, no timeout on FFI without kill-task strategy, no `DashMap` guard across `.await`, duplicate open ID returns `-1`. Simplify: duplicated consume/forward paths, label lookups per message, re-expanded log-layer matches.