test: rejig correctness test

[quettabit]

No description provided.

[greptile-apps]

Greptile Summary

This PR refactors the correctness test to use record headers instead of the body for carrying an index, removes the ReadLimit in favour of an in-loop early return, and improves error reporting with pytest.fail.

• Record identity moves from the body (plain integer string) to a record-idx header, making the body semantically neutral and allowing future headers without breaking index parsing.
• The read session now runs without a count limit; termination is driven by next_record_idx reaching _NUM_RECORDS, with a pytest.fail fallback when the session expires via the 60-second wait timeout.
• The append_records task drops the ack.seq_num >= 0 assertion (trivially always true) and uses _indexed_record / _record_idx helpers for cleaner index handling.

Confidence Score: 5/5

Safe to merge; the refactoring is logically sound and the intentional design choices around duplicate tolerance are consistent with the developer's stated goals.

The logic changes are well-reasoned: seq-num contiguity is correctly enforced, the in-loop early return correctly terminates when all distinct record indices are observed, and the pytest.fail fallback cleanly handles session expiry. The termination assertions are intentionally permissive to tolerate retry-induced duplicates, a design choice already discussed and accepted in the prior review thread.

No files require special attention; the single changed file is a test helper with no production impact.

Important Files Changed

Filename	Overview
tests/test_correctness.py	Refactors correctness test: switches record identity to headers, removes ReadLimit, tightens early-return logic, and adds pytest.fail fallback; termination assertions are intentionally permissive to tolerate retry-induced duplicates.

Sequence Diagram

sequenceDiagram
    participant TG as TaskGroup
    participant A as append_records
    participant S as S2Stream
    participant R as read_records

    TG->>A: create_task
    TG->>R: create_task

    loop for idx in 0.._NUM_RECORDS-1
        A->>S: producer.submit(_indexed_record(idx))
        S-->>A: ticket
    end
    loop await tickets
        A->>S: await ticket
        S-->>A: ack
    end

    S-->>R: "read_session(start=SeqNum(0), wait=60)"
    loop for each batch
        loop for each record
            R->>R: assert seq_num contiguous
            R->>R: "assert record_idx <= next_record_idx"
            R->>R: advance next_record_idx if at frontier
            alt "next_record_idx == _NUM_RECORDS"
                R->>R: "assert last_seq_num + 1 == num_records_read"
                R->>R: "assert num_records_read >= _NUM_RECORDS"
                R-->>TG: return (success)
            end
        end
    end
    R->>TG: pytest.fail (session ended before all records seen)

Loading

Prompt To Fix All With AI

Fix the following 1 code review issue. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 1
tests/test_correctness.py:82-85
When `_record_idx` is called on a record that has no `record-idx` header (e.g., if the stream already contains records from a previous test run that were appended without a header, or if a retry surfaces a record from outside this test), the `assert len(values) == 1` fires with `len(values) == 0`. This produces a bare `AssertionError` rather than a descriptive failure, making it harder to diagnose the root cause. Adding an explicit message makes the failure immediately actionable.

```suggestion
def _record_idx(record: SequencedRecord) -> int:
    values = [value for key, value in record.headers if key == _RECORD_IDX_HEADER]
    assert len(values) == 1, (
        f"expected exactly one {_RECORD_IDX_HEADER!r} header, "
        f"got {len(values)} on seq_num={record.seq_num}"
    )
    return int(values[0])
```

_{Reviews (2): Last reviewed commit: "initial commit" | Re-trigger Greptile}

[quettabit]

@greptileai addressed your comment. pls review again.