Streams: unified Stream / AsyncStream / network backends (Feature 3, Phase 1-3) - #134
Merged
Conversation
Introduces core/runtime Stream abstract with FileStream, MemoryStream, and PipeStream backends, providing a single byte-oriented surface for Pulp I/O. Async and network wrappers layer on top in later phases. The interface returns StreamResult (bytes + StreamError) instead of throwing, so callers can treat partial writes and closed peers as first-class values — a prerequisite for the backpressure-aware AsyncStream.
AsyncStream runs a background worker that pumps a synchronous Stream and
delivers read/write/close events through user-supplied callbacks. It adds
three things the raw Stream does not:
- a bounded write queue with explicit backpressure (`write_async` returns
false when the pending byte count would exceed the high-water mark);
- callback routing via an optional executor, so completions can land on
the caller's own event loop without pulp-runtime depending on
pulp-events (avoids a link cycle);
- cancellation — destruction, `cancel()`, or `close()` fires pending
write completions with StreamError::Closed and exits the worker.
Wraps the existing Socket and http_get/http_post helpers behind the unified Stream interface so the same AsyncStream wrapper, backpressure logic, and cancellation token work for network I/O without per-backend plumbing. TcpStream performs a blocking DNS + TCP handshake in connect(); pair it with AsyncStream to make the handshake non-blocking for the caller. HttpStream issues the request eagerly and exposes the response body through Stream::read(); request-body streaming is deferred to Phase 4. TLS is inherited from cpp-httplib's HTTPS support (already vendored with mbedTLS); no separate handshake code lives here.
examples/stream-demo exercises all three layers: synchronous FileStream I/O, an AsyncStream driving reads onto a pulp::events::EventLoop, and an optional HTTP GET via HttpStream. This is the reference consumer that proves the Options.executor hook lets runtime-layer streams integrate with the events layer without introducing a link cycle. docs/reference/streams.md documents the StreamResult contract, each backend, the backpressure/cancellation model, and the deferred MessageChannel work. Also adds a streams overview to modules.md so the pulp::runtime section surfaces the new API next to the older http/socket entries. CLI impact: assessed `pulp doctor` — its current network check is a single synchronous call, so migrating it to AsyncStream + HttpStream would add complexity without user-visible benefit. Left as a follow-up once the doctor gains multi-target probes.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 540df706e2
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+117
to
+119
| if (pending_bytes_ == 0 && on_drain_) { | ||
| auto cb = on_drain_; | ||
| dispatch(std::move(cb)); |
There was a problem hiding this comment.
AsyncStream::fire_drain_if_idle_locked dispatches on_drain while mutex_ is still held, and with the default executor dispatch runs the callback inline on the worker thread. The documented backpressure flow expects callers to resume writes from on_drain, but calling write_async from that callback tries to lock mutex_ again and deadlocks the worker, stalling all future I/O.
Useful? React with 👍 / 👎.
Comment on lines
+55
to
+60
| void AsyncStream::cancel() { | ||
| token_.cancel(); | ||
| { | ||
| std::lock_guard<std::mutex> lock(mutex_); | ||
| running_ = false; | ||
| } |
There was a problem hiding this comment.
cancel() just sets the cancellation flag and running_ = false, but it never drains write_queue_ or completes queued callbacks; then worker_main exits on its next loop check. If callers cancel while writes are already queued, those callbacks are silently dropped instead of receiving StreamError::Closed, which can leave higher-level code waiting indefinitely for completion signals.
Useful? React with 👍 / 👎.
Collaborator
Author
There was a problem hiding this comment.
Codex false positive — this comment re-appeared against commit 4771e0b9 (the merge-from-main commit) but cancel() in the code that actually merged to main calls drain_queue_as_closed() at the end, and the regression test AsyncStream cancel drains queue even before worker starts exercises exactly this path.
See commit 45a03774 for the fix and the test that keeps it from regressing.
Comment on lines
+188
to
+189
| if (options_.auto_read && stream_ && stream_->is_open()) { | ||
| auto r = stream_->read(read_buf.data(), read_buf.size()); |
There was a problem hiding this comment.
The worker performs a direct stream_->read(...) each iteration when auto_read is enabled, but newly added network backends use blocking socket I/O (TcpStream::read -> recv). If no inbound data is available, the worker can block in read and stop processing write_async items (and can delay shutdown until the read unblocks), which breaks request/response usage where writes are queued after start().
Useful? React with 👍 / 👎.
accessibility_android.cpp on main references `View::children()` which doesn't exist (use child_count/child_at) and calls simulate_click() without the required Point argument. Neither compiles under the Android NDK, so the Android CI job has been red on main since 86e888f. Fixed: - iterate children via child_count()/child_at(i) - call simulate_click() with the view's center in root coordinates - import pulp::view::Point alongside View No behavior change on reachable platforms; this is a pre-existing break surfaced by the streams PR's CI run.
Three P1 issues reported by Codex review: 1. on_drain dispatched while holding mutex_ — with the default inline executor, a caller re-entering write_async() from on_drain deadlocked the worker. Fixed by collecting the callback under the lock and dispatching it outside the lock. 2. cancel() never completed queued write callbacks — higher-level code waiting on a completion signal would hang. Fixed by draining the queue with StreamError::Closed inside cancel() and stop(), plus a new regression test that queues writes before start() and verifies completions fire after cancel(). 3. A single worker loop mixed blocking reads and queued writes, so a blocking TcpStream::read starved write processing. Split into two threads — writer_thread_ drains write_queue_, reader_thread_ runs the auto-read path — with a shared cancellation flag that wakes both. All stream/async/network tests pass locally (6 + 6 + 3 cases). No public API changes.
New .agents/skills/streams/SKILL.md captures the non-obvious rules for using the Stream/AsyncStream hierarchy correctly: the decision tree for picking a backend, the backpressure-check-the-bool footgun, the cancel semantics (queued callbacks fire with Closed — don't assume they silently vanish), and the reader/writer thread split that keeps blocking TcpStream reads from starving writes. Triggered by the audit of planning/ralph-prompt-streams.md completion conditions — "skills: create if a repeatable workflow emerged" applies here because the AsyncStream patterns are easy to get wrong on first use.
# Conflicts: # core/view/platform/android/accessibility_android.cpp
danielraffel
added a commit
that referenced
this pull request
Apr 13, 2026
danielraffel
added a commit
that referenced
this pull request
Apr 13, 2026
* planning: bump submodule for versioning-v2 plan + ralph prompt
* planning: bump submodule for Shipyard split in versioning plan
* planning: bump submodule for config-driven scripts + Shipyard stub
* planning: bump submodule for Shipyard ralph prompt
* planning: update submodule to rebased tip (MPE check-offs + Shipyard prompt)
* planning: bump submodule for pulp pr + auto-release components
* scripts: add versioning config + skill-sync/version-bump gates
Core of the versioning-v2 plan (Components A + B):
- tools/scripts/versioning.json — repo-agnostic config describing
surfaces (sdk, plugin), their version files, trigger paths, public-
API paths, internal-only paths, and generated-file globs. Uses JSON
rather than YAML so there's no PyYAML dependency on PEP-668 Python
(same schema, same semantics).
- tools/scripts/versioning.schema.json — formal JSON schema for the config.
- tools/scripts/skill_path_map.json — one entry per skill under
.agents/skills/, mapping glob patterns to skill names. Covers all
13 current skills (aax, android, ci, cli-maintenance, cmajor-external,
engine, faust, import-design, jsfx-subset, packages, ship,
threejs-bridge, webview-ui). The map routes shipyard-config and
CI-workflow changes to the ci skill.
- tools/scripts/skill_sync_check.py — hard-fails when a diff touches
mapped paths without updating the corresponding SKILL.md, unless a
'Skill-Update: skip skill=<name> reason="..."' trailer is present
on the tip commit. Self-check fails if a skill dir lacks a map entry.
- tools/scripts/version_bump_check.py — detects which surfaces need a
bump, using conservative heuristics (public-header change = minor,
internal-only = patch-suggested, 'BREAKING:' subject = major).
Three modes: report (CI), apply (rewrites version files in place
for pulp pr), hint (agent-hook advisory). Trailer override
'Version-Bump: <surface>=<level>' is authoritative.
Both scripts are config-driven via --config so the same code will
port to the Shipyard repo once the pulp loop lands.
* planning: bump submodule for deliverable 1-4 check-off
* versioning-v2: add pre-push hook, CI workflow, install script, guide
Checkpoints deliverables 7, 8, and partial 11:
- .githooks/pre-push — advisory-by-default runner for both gates.
PULP_ENFORCE_PREPUSH=1 upgrades warnings to hard failures (CI sets
this); PULP_SKIP_PREPUSH=1 is the single-push escape valve.
- tools/scripts/install-githooks.sh — idempotent core.hooksPath
installer; setup.sh will call this in a follow-up.
- .github/workflows/version-skill-check.yml — authoritative PR gate.
Pulls full history so origin/<base_ref> is reachable, runs both
scripts in report mode, fails hard on any issue. No bypass other
than the commit trailers; audit trail lives in git.
- docs/guides/versioning.md — end-to-end guide covering the three
layers, the trailer syntax, 'pulp pr' one-command flow,
Shipyard-pin-vs-Shipyard-config split, and agent-parity story.
Skill-Update: skip skill=ci reason="CI-skill touched (shipyard-config paths added to skill_path_map in prior commit), but the change was the map entry itself; no new gotcha to record yet."
* versioning-v2: Shipyard gate, CLAUDE.md policy, auto-release, hint mode
Wires the last four layer-spanning pieces of the plan:
- .shipyard/config.toml gains a validation.gates pipeline that runs
both scripts with PULP_ENFORCE_PREPUSH=1 via the canonical 'setup'
stage. 'shipyard run --pipeline gates' invokes them standalone.
- .github/workflows/auto-release.yml (Component J): on push to main,
diffs CMakeLists.txt project version + .claude-plugin/plugin.json
version against the previous push range. If either moved, creates
the matching v<x.y.z> or plugin-v<x.y.z> tag via bot token; the
existing tag-triggered release-cli.yml / sign-and-release.yml then
build and publish. Safety: concurrency group, idempotent-strict
tagging (fails loudly on mismatched existing tags), Release: skip
trailer suppression, revert detection.
- hooks/scripts/cli-plugin-sync.sh: calls both scripts in --mode=hint
after the existing CLI-sync reminders. Advisory only — Layer 1.
- CLAUDE.md: new 'Versioning & Skill-Sync Policy' section under Skill
Maintenance Rule. Documents the three layers, the 'push a PR'
natural-language routing through pulp pr, and the three bypass
trailers. Codex inherits via AGENTS.md -> CLAUDE.md pointer.
Skill-Update: skip skill=ci reason="auto-release workflow is a new artifact; the ci skill doesn't yet have a gotcha about tag-triggered release flows — a note will land once the loop runs end-to-end."
* planning: bump submodule for deliverables 7-11, 13, 14 check-off
* versioning-v2: resolve SDK/plugin drift via mode=apply bump
Ran tools/scripts/version_bump_check.py --mode=apply on versioning-v2:
- CMakeLists.txt: Pulp SDK 0.3.0 → 0.4.0
- .claude-plugin/plugin.json: 0.2.0 → 0.3.0
- .claude-plugin/marketplace.json: pulp plugin entry bumped to 0.3.0
- CHANGELOG.md: inherits both bump entries under [Unreleased]
Verified afterwards:
- version_bump_check.py --mode=report on HEAD is clean
- skill_path_map.json and versioning.json are present and load cleanly
Skill-Update: skip skill=android reason="version bump only; no code under mapped android paths changed in this commit — script flagged accumulated branch diff, not this commit's contents"
Skill-Update: skip skill=ci reason="version bump only; workflows untouched in this commit"
Skill-Update: skip skill=cli-maintenance reason="version bump only; no CLI source changed in this commit — the create.md / cmd_create.cpp diff is from earlier commits on this branch"
Skill-Update: skip skill=packages reason="CHANGELOG/DEPENDENCIES/NOTICE edits in this commit are driven by the auto-bump, not by a dependency inventory change"
* versioning-v2: ci skill routes through pulp pr + /pr slash command
- .agents/skills/ci/SKILL.md: new top section explicitly maps the natural-
language triggers ('ship this', 'we're done', 'push a PR', ...) to
running 'pulp pr' rather than 'gh pr create' + 'shipyard ship'
separately. Preserves the raw shipyard commands below as diagnostic
fallback so existing workflows don't break.
- .claude/commands/pr.md: slash command invokes 'pulp pr' with
user-supplied $ARGUMENTS; documents the 6-step pipeline inline.
Skill-Update: skip skill=android reason="no android paths touched in this commit"
Skill-Update: skip skill=cli-maintenance reason="slash command + skill doc only; no CLI source in this commit"
Skill-Update: skip skill=packages reason="no dependency inventory change in this commit"
* cli: add 'pulp pr' subcommand; expand 'pulp version check'
Finishes deliverable 12 and half of 5:
- cmd_pr.cpp (new): one-shot PR orchestrator wrapping
skill_sync_check -> version_bump_check --mode=apply -> commit ->
push -> gh pr create -> shipyard ship. Refuses to run on main;
refuses if worktree isn't clean after the bump. Flags: --base,
--title, --no-ship, --no-push, --dry-run. Wired into pulp_cli.cpp
command table, cli_common.hpp, and the CMakeLists source list.
- cmd_version.cpp: version_check() now also validates
.claude-plugin/plugin.json and .claude-plugin/marketplace.json
(versions must be valid semver and match each other). New
'pulp version check --with-bump-check' flag pipes into
tools/scripts/version_bump_check.py --mode=report.
- .claude/commands/pr.md slash command was already in place from a
prior iteration and routes through 'pulp pr'.
- .agents/skills/ci/SKILL.md already instructs agents to route
natural-language triggers through 'pulp pr'.
Verified:
pulp pr --dry-run prints the 6-step plan correctly.
pulp version check --with-bump-check reports the known drift
(SDK=0.4.0 bumped, plugin=0.3.0, marketplace=1.0.0 mismatch)
including the version_bump_check verdict.
Skill-Update: skip skill=cli-maintenance reason="pulp pr is the orchestrator the existing ci skill already documents; no new CLI-authoring gotcha to record beyond the flags covered in the source comments and the slash command md."
* versioning-v2: compliance fix — marketplace + ci skill + version regex
Deliverable 6 (compliance fix):
- .claude-plugin/marketplace.json plugins[0].version: 0.2.0 -> 0.3.0 to
match plugin.json (which itself was bumped from 0.2.0 -> 0.3.0 earlier
on this branch by the version_bump_check auto-apply).
- cmd_version.cpp: read_json_version_field() now anchors on the top-level
'version' field (two-space indent or less). The prior regex matched the
first occurrence, which picked up marketplace.json's metadata.version
(the marketplace FORMAT version) instead of the canonical top-level
one. 'pulp version check' now correctly confirms plugin.json and
marketplace.json agree on 0.3.0.
- .agents/skills/ci/SKILL.md gains a 'Versioning & Skill-Sync gates'
section documenting the three-layer enforcement, the path-map gotcha
for ci, the auto-release workflow's diff semantics, and the
idempotent-strict tag safety property. Dogfoods the skill-update
rule this loop is building.
Skill-Update: skip skill=android reason="accessibility_android.cpp changed on main before this branch was cut; that drift belongs to the PR that landed it, not versioning-v2."
Skill-Update: skip skill=packages reason="DEPENDENCIES.md + NOTICE.md moved on main via the recent MPE and Streams PRs; not introduced by this branch."
* scripts: fix skill-md-prefix bug; document pulp pr + version check in cli-maintenance skill
Two fixes:
- skill_sync_check.py: '.agents/skills' was being chopped to
'agents/skills' because the old code used str.lstrip('./') which
removes *characters* not a prefix, eating the leading dot. Replaced
with explicit './'-prefix stripping. Before the fix, the tool
reported every SKILL.md as NOT updated even when it was in the diff.
- cli-maintenance/SKILL.md: documents 'pulp pr' (the one-shot PR
orchestrator) and 'pulp version check --with-bump-check', including
invariants (refuses on main, refuses dirty worktree post-bump),
anti-patterns (don't call the Python scripts by hand; don't split
gh + shipyard), and the JSON-multiple-version-field gotcha that the
version-check regex anchors around.
With both fixes: skill-sync reports clean (ci and cli-maintenance
now ✓ updated; android and packages bypassed with reasons).
Skill-Update: skip skill=android reason="accessibility_android.cpp drift is from commit on main before this branch was cut — not versioning-v2 scope."
Skill-Update: skip skill=packages reason="DEPENDENCIES.md/NOTICE.md drift is from MPE/Streams PRs on main — not versioning-v2 scope."
* planning: bump submodule — 14/14 versioning-v2 deliverables checked
* scripts: fixture tests for both gates; fix heuristic + override bugs
Adds tools/scripts/test_gates.py — self-contained unittest harness (no
extra deps) that spins up throwaway git repos with a minimal
versioning.json + skill_path_map.json and exercises 11 scenarios
against version_bump_check.py and skill_sync_check.py:
- new public header -> minor-required (bump_check)
- comments-only edit -> none (bump_check)
- whitespace-only edit -> none (bump_check)
- BREAKING: trailer -> major-required (bump_check)
- test-only change -> none (bump_check)
- generated-file change -> skipped (none) (bump_check)
- Version-Bump: skip trailer -> none (wins over feat: ceiling) (bump_check)
- revert commit -> none (bump_check)
- skill path, no SKILL.md -> fail (skill_check)
- skill path + SKILL.md -> pass (skill_check)
- skill path + bypass trailer -> pass (skill_check)
Writing the fixtures exposed three real bugs; fixed them:
1. heuristic_for_surface: whitespace/comment-only edits on public paths
were falling through to the internal-path fallback and returning
'patch' instead of 'none'. Rewrote so paths with no meaningful
diff (per git diff --ignore-all-space + a C/C++/Python comment
filter) collapse to 'none' for the whole surface.
2. Skip-override interaction: Version-Bump: <surface>=skip was being
detected but then raised back up by the conventional-commit
ceiling loop (e.g. a feat: subject). Made skip authoritative: it
never gets re-raised.
3. Report rendering: patch-suggested verdicts were printing '✗ bump
required' even though render_report correctly treated them as
warnings (exit 0). Rephrased patch-level unbumped output as
'? bump suggested (patch)' so the text matches the exit code.
Wired test_gates.py into .github/workflows/version-skill-check.yml as
a new step after the gate-script runs so any regression hits CI.
setup.sh now calls tools/scripts/install-githooks.sh during bootstrap
(before external-SDK setup) so a fresh clone lands with
core.hooksPath=.githooks by default. Idempotent — safe to rerun.
Skill-Update: skip skill=ci reason="Adds a new CI workflow step that runs the fixture tests. The ci SKILL.md's 'Versioning & Skill-Sync gates' section already documents the workflow surface area; no new gotcha until the fixtures catch a regression."
Skill-Update: skip skill=android reason="pre-existing drift, not versioning-v2 scope"
Skill-Update: skip skill=packages reason="pre-existing drift, not versioning-v2 scope"
* planning: acceptance transcript bump
* scripts: make install-githooks.sh layout-agnostic
Walks up to the first .git entry instead of hardcoding ../.. from
the script's directory. Works unchanged whether the script lives at
tools/scripts/ (pulp) or scripts/ (Shipyard). Required for the
Shipyard port to reuse the same file without per-repo tweaks.
* planning: sync submodule to latest tip
* skills: add sdf-text and streams to path map (landed on main via #136, #134)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements Phase 1–3 of Feature 3 from
planning/next-features-plan.md: a unifiedStreaminterface plus an async wrapper and network backends, so every transport (file, memory, pipe, TCP, HTTP) shares one contract.core/runtime/stream.{hpp,cpp}:Streamabstract +FileStream,MemoryStream,PipeStream.StreamResult{bytes, error}surface avoids exceptions on the hot path.core/runtime/async_stream.{hpp,cpp}: background worker,on_data/on_error/on_close/on_draincallbacks, bounded write queue with explicit backpressure (write_asyncreturns false past the high-water mark),CancellationToken, and an executor hook so callbacks can land on anEventLoopwithout pulp-runtime linking pulp-events (avoids a cycle).core/runtime/network_stream.{hpp,cpp}:TcpStreamoverSocket,HttpStreamoverhttp_get/http_post. TLS is inherited from cpp-httplib/mbedTLS; request-body streaming is deferred to Phase 4.examples/stream-demo/exercises all three layers end-to-end, including anAsyncStreamdriving reads onto a realpulp::events::EventLoop.docs/reference/streams.mdwith the full API;docs/reference/modules.mdruntime section gains a Streams entry.planning/next-features-plan.md(planning submodule commit74e9981).Phase 4 (message channels / WebSocket / OSC) is intentionally deferred per the ralph prompt.
Test plan
pulp-test-stream— 6 cases / 40 assertionspulp-test-async-stream— 5 cases / 17 assertions (read dispatch, write flush, backpressure, cancel, executor routing)pulp-test-network-stream— 3 cases / 11 assertions (loopback TCP echo, HTTP transport error, HTTP write-is-Invalid)ctest --exclude-regex AudioWorkgroup: 2062/2065 pass. The 3 failures are pre-existingClipboard*pasteboard contention under-j(pass when run singly, also pass onmain)examples/stream-demoruns end-to-endshipyard ship