Implement app streams: stdio for apps (#640)

Implement file IO for applications to facilitate text input/output capturing.

Apps can now launch apps and:
- Read their stdio output
- Write to their stdio input

Modules/app-module/source/fd_table.cpp

@@ -0,0 +1,151 @@
+// SPDX-License-Identifier: Apache-2.0
+#include <app/private/fd_table.h>
+#include <app/private/null_device.h>
+
+#include <tactility/check.h>
+#include <tactility/concurrent/mutex.h>
+#include <tactility/error.h>
+
+namespace {
+
+bool fd_in_range(int fd) {
+    return fd >= 0 && fd < APP_MAX_FDS;
+}
+
+bool get_internal(AppFdTable* table, int fd, AppFile* out_file, bool retain) {
+    if (!fd_in_range(fd)) {
+        return false;
+    }
+
+    mutex_lock(&table->mutex);
+    check(!table->shutting_down);
+    AppFdSlot& slot = table->slots[fd];
+    bool in_use = slot.in_use;
+    if (in_use) {
+        *out_file = slot.file;
+        if (retain && out_file->ops->retain != nullptr) {
+            out_file->ops->retain(out_file->object);
+        }
+    }
+    mutex_unlock(&table->mutex);
+    return in_use;
+}
+
+} // namespace
+
+extern "C" {
+
+void app_fd_table_construct(AppFdTable* table) {
+    const AppFile* null_file = app_null_file();
+    for (int fd = 0; fd < APP_MAX_FDS; fd++) {
+        table->slots[fd].in_use = false;
+        table->slots[fd].ever_used = false;
+    }
+    for (int fd = STDIN_FILENO; fd <= STDERR_FILENO; fd++) {
+        table->slots[fd].file = *null_file;
+        table->slots[fd].in_use = true;
+        table->slots[fd].ever_used = true;
+    }
+    table->shutting_down = false;
+    mutex_construct(&table->mutex);
+}
+
+void app_fd_table_teardown(AppFdTable* table) {
+    AppFile closing[APP_MAX_FDS];
+    int closing_count = 0;
+
+    mutex_lock(&table->mutex);
+    table->shutting_down = true;
+    for (int fd = 0; fd < APP_MAX_FDS; fd++) {
+        if (table->slots[fd].in_use) {
+            closing[closing_count++] = table->slots[fd].file;
+            table->slots[fd].in_use = false;
+        }
+    }
+    mutex_unlock(&table->mutex);
+
+    for (int i = 0; i < closing_count; i++) {
+        closing[i].ops->close(closing[i].object);
+    }
+
+    mutex_destruct(&table->mutex);
+}
+
+error_t app_fd_table_bind(AppFdTable* table, int fd, const AppFileOps* ops, void* object) {
+    if (!fd_in_range(fd)) {
+        return ERROR_OUT_OF_RANGE;
+    }
+
+    mutex_lock(&table->mutex);
+    check(!table->shutting_down);
+    AppFdSlot& slot = table->slots[fd];
+    AppFile old = slot.file;
+    bool had_old = slot.in_use;
+    slot.file = { .ops = ops, .object = object };
+    slot.in_use = true;
+    slot.ever_used = true;
+    mutex_unlock(&table->mutex);
+
+    if (had_old) {
+        old.ops->close(old.object);
+    }
+    return ERROR_NONE;
+}
+
+error_t app_fd_table_allocate(AppFdTable* table, const AppFileOps* ops, void* object, int* out_fd) {
+    mutex_lock(&table->mutex);
+    check(!table->shutting_down);
+    for (int fd = 3; fd < APP_MAX_FDS; fd++) {
+        AppFdSlot& slot = table->slots[fd];
+        if (!slot.in_use) {
+            slot.file = { .ops = ops, .object = object };
+            slot.in_use = true;
+            slot.ever_used = true;
+            mutex_unlock(&table->mutex);
+            *out_fd = fd;
+            return ERROR_NONE;
+        }
+    }
+    mutex_unlock(&table->mutex);
+    return ERROR_RESOURCE;
+}
+
+bool app_fd_table_get(AppFdTable* table, int fd, AppFile* out_file) {
+    return get_internal(table, fd, out_file, /*retain=*/false);
+}
+
+bool app_fd_table_get_and_retain(AppFdTable* table, int fd, AppFile* out_file) {
+    return get_internal(table, fd, out_file, /*retain=*/true);
+}
+
+error_t app_fd_table_close(AppFdTable* table, int fd) {
+    if (!fd_in_range(fd)) {
+        return ERROR_NOT_FOUND;
+    }
+
+    mutex_lock(&table->mutex);
+    check(!table->shutting_down);
+    AppFdSlot& slot = table->slots[fd];
+    bool in_use = slot.in_use;
+    AppFile old = slot.file;
+    slot.in_use = false;
+    mutex_unlock(&table->mutex);
+
+    if (!in_use) {
+        return ERROR_NOT_FOUND;
+    }
+    return old.ops->close(old.object);
+}
+
+bool app_fd_table_is_app_owned(AppFdTable* table, int fd) {
+    if (!fd_in_range(fd)) {
+        return false;
+    }
+
+    mutex_lock(&table->mutex);
+    bool owned = table->slots[fd].in_use || table->slots[fd].ever_used;
+    mutex_unlock(&table->mutex);
+    return owned;
+}
+
+} // extern "C"

Modules/app-module/source/io.cpp

@@ -0,0 +1,106 @@
+// SPDX-License-Identifier: Apache-2.0
+#include <app/io.h>
+
+#include <app/private/fd_table.h>
+#include <app/private/ledger.h>
+#include <app/scheduler.h>
+
+#include <cerrno>
+
+#ifdef ESP_PLATFORM
+extern "C" {
+ssize_t __real_read(int fd, void* buffer, size_t size);
+ssize_t __real_write(int fd, const void* buffer, size_t size);
+int __real_close(int fd);
+}
+#else
+#include <unistd.h>
+#endif
+
+namespace {
+
+// NULL for a caller not running as an app instance (e.g. a kernel service task). @a fd is then
+// a real underlying fd, handled by the caller falling through to the real syscall.
+AppFdTable* current_app_fd_table() {
+    AppInstanceId app_id = app_scheduler_current_app_id();
+    if (app_id == 0) {
+        return nullptr;
+    }
+    auto& ledger = app_ledger();
+    mutex_lock(&ledger.mutex);
+    auto iterator = ledger.instances.find(app_id);
+    AppFdTable* table = (iterator != ledger.instances.end()) ? &iterator->second.fd_table : nullptr;
+    mutex_unlock(&ledger.mutex);
+    return table;
+}
+
+} // namespace
+
+extern "C" {
+
+ssize_t app_io_read(int fd, void* buffer, size_t size) {
+    AppFdTable* table = current_app_fd_table();
+    AppFile file {};
+    if (table != nullptr && app_fd_table_get_and_retain(table, fd, &file)) {
+        ssize_t result = file.ops->read(file.object, buffer, size);
+        if (file.ops->release != nullptr) {
+            file.ops->release(file.object);
+        }
+        return result;
+    }
+    // @a fd isn't currently bound. If this table has never touched it either, it's a real
+    // underlying fd (e.g. from fopen()/open(), which app-module never intercepts; see app/io.h),
+    // so fall through. Otherwise it's one of ours that's already been closed: a real EBADF, not
+    // a real fd to hand to the platform (which could belong to something else entirely by now).
+    if (table != nullptr && app_fd_table_is_app_owned(table, fd)) {
+        errno = EBADF;
+        return -1;
+    }
+#ifdef ESP_PLATFORM
+    return __real_read(fd, buffer, size);
+#else
+    return ::read(fd, buffer, size);
+#endif
+}
+
+ssize_t app_io_write(int fd, const void* buffer, size_t size) {
+    AppFdTable* table = current_app_fd_table();
+    AppFile file {};
+    if (table != nullptr && app_fd_table_get_and_retain(table, fd, &file)) {
+        ssize_t result = file.ops->write(file.object, buffer, size);
+        if (file.ops->release != nullptr) {
+            file.ops->release(file.object);
+        }
+        return result;
+    }
+    if (table != nullptr && app_fd_table_is_app_owned(table, fd)) {
+        errno = EBADF;
+        return -1;
+    }
+#ifdef ESP_PLATFORM
+    return __real_write(fd, buffer, size);
+#else
+    return ::write(fd, buffer, size);
+#endif
+}
+
+int app_io_close(int fd) {
+    AppFdTable* table = current_app_fd_table();
+    if (table != nullptr) {
+        error_t result = app_fd_table_close(table, fd);
+        if (result == ERROR_NONE) {
+            return 0;
+        }
+        if (app_fd_table_is_app_owned(table, fd)) {
+            errno = EBADF;
+            return -1;
+        }
+    }
+#ifdef ESP_PLATFORM
+    return __real_close(fd);
+#else
+    return ::close(fd);
+#endif
+}
+
+} // extern "C"

Modules/app-module/source/manager.cpp

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 #include <app/manager.h>
 #include <app/metadata.h>
+#include <app/private/fd_table.h>
 #include <app/private/fs.h>
 #include <app/private/ledger.h>
 #include <app/private/scheduler.h>
@@ -89,8 +90,15 @@ char** copy_arguments(int argc, const char* const argv[]) {
 
 // Takes ownership of argv (already a deep copy, or NULL/argc==0) regardless of outcome -
 // app_scheduler_start() frees it on any failure path, and the spawned task frees it once its
-// run() returns.
-error_t start_internal(const char* id, AppInstanceId parent_instance_id, int argc, char* argv[], AppInstanceId* out_app_instance_id) {
+// run() returns. @a bindings (@a binding_count entries, may be NULL/0) are subscribed into the
+// new instance's fd table before app_scheduler_start() is called, so they're in place before its
+// task begins executing (see app_manager_start_with_streams()).
+error_t start_internal(const char* id, AppInstanceId parent_instance_id, int argc, char* argv[], const AppStreamBinding* bindings, size_t binding_count, AppInstanceId* out_app_instance_id) {
+    if (binding_count != 0 && bindings == nullptr) {
+        app_ledger_free_arguments(argc, argv);
+        return ERROR_INVALID_ARGUMENT;
+    }
+
     auto& ledger = app_ledger();
 
     mutex_lock(&ledger.mutex);
@@ -106,13 +114,42 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg
     AppInstanceRecord record { .id = target_id, .manifest = manifest, .state = APP_INSTANCE_STATE_STARTING, .task = nullptr };
     record.parent_id = parent_instance_id;
     ledger.instances[target_id] = record;
+    // Constructed on the map-resident copy, not the local `record` about to go out of scope.
+    // AppFdTable::fds[] entries point into AppFdTable::slots[] by address (see fd_table.h), so
+    // constructing before the copy above would leave them pointing at stack storage.
+    app_fd_table_construct(&ledger.instances[target_id].fd_table);
     mutex_unlock(&ledger.mutex);
 
     LOG_I(TAG, "[instance %d] starting %s with parent %d", target_id, manifest->id, parent_instance_id);
 
+    for (size_t i = 0; i < binding_count; i++) {
+        error_t bind_result = app_stream_subscribe(bindings[i].stream, bindings[i].buffer, bindings[i].buffer_capacity, bindings[i].event_group, target_id, bindings[i].producer_fd);
+        if (bind_result != ERROR_NONE) {
+            LOG_E(TAG, "[instance %d] Failed to bind stream at fd %d: %s", target_id, bindings[i].producer_fd, error_to_string(bind_result));
+            // Undo bindings[0..i): app_fd_table_teardown() below only closes each stream. It
+            // doesn't release the event bits app_stream_subscribe() claimed or destruct
+            // stream->internal.mutex; only app_stream_unsubscribe() does that.
+            for (size_t j = 0; j < i; j++) {
+                app_stream_unsubscribe(bindings[j].stream);
+            }
+            mutex_lock(&ledger.mutex);
+            app_fd_table_teardown(&ledger.instances[target_id].fd_table);
+            ledger.instances.erase(target_id);
+            mutex_unlock(&ledger.mutex);
+            app_ledger_free_arguments(argc, argv);
+            return bind_result;
+        }
+    }
+
     error_t error = app_scheduler_start(target_id, manifest->location, manifest->stack, argc, argv);
     if (error != ERROR_NONE) {
+        // Every binding succeeded before app_scheduler_start() failed. Unsubscribe all of them,
+        // same reasoning as the bind-failure path above.
+        for (size_t j = 0; j < binding_count; j++) {
+            app_stream_unsubscribe(bindings[j].stream);
+        }
         mutex_lock(&ledger.mutex);
+        app_fd_table_teardown(&ledger.instances[target_id].fd_table);
         ledger.instances.erase(target_id);
         mutex_unlock(&ledger.mutex);
         LOG_I(TAG, "[instance %d] Failed to start: %s", target_id, error_to_string(error));
@@ -126,15 +163,19 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg
 } // namespace
 
 error_t app_manager_start(const char* id, AppInstanceId* out_app_instance_id) {
-    return start_internal(id, 0, 0, nullptr, out_app_instance_id);
+    return start_internal(id, 0, 0, nullptr, nullptr, 0, out_app_instance_id);
 }
 
 error_t app_manager_start_with_parameters(const char* id, int argc, const char* const argv[], AppInstanceId* out_app_instance_id) {
-    return start_internal(id, 0, argc, copy_arguments(argc, argv), out_app_instance_id);
+    return start_internal(id, 0, argc, copy_arguments(argc, argv), nullptr, 0, out_app_instance_id);
 }
 
 error_t app_manager_start_for_result(const char* id, AppInstanceId parent_instance_id, int argc, const char* const argv[], AppInstanceId* out_app_instance_id) {
-    return start_internal(id, parent_instance_id, argc, copy_arguments(argc, argv), out_app_instance_id);
+    return start_internal(id, parent_instance_id, argc, copy_arguments(argc, argv), nullptr, 0, out_app_instance_id);
+}
+
+error_t app_manager_start_with_streams(const char* id, const AppStreamBinding* bindings, size_t binding_count, AppInstanceId* out_app_instance_id) {
+    return start_internal(id, 0, 0, nullptr, bindings, binding_count, out_app_instance_id);
 }
 
 error_t app_manager_stop(AppInstanceId app_instance_id) {

Modules/app-module/source/module.cpp

@@ -1,11 +1,13 @@
 // SPDX-License-Identifier: Apache-2.0
 #include <app/event.h>
 #include <app/install.h>
+#include <app/io.h>
 #include <app/manager.h>
 #include <app/manifest.h>
 #include <app/metadata.h>
 #include <app/paths.h>
 #include <app/scheduler.h>
+#include <app/stream.h>
 
 #include <service/manager.h>
 
@@ -27,10 +29,15 @@ static const ModuleSymbol SYMBOLS[] = {
     DEFINE_MODULE_SYMBOL(app_get_install_path),
     DEFINE_MODULE_SYMBOL(app_install),
     DEFINE_MODULE_SYMBOL(app_uninstall),
+    // app/io
+    DEFINE_MODULE_SYMBOL(app_io_read),
+    DEFINE_MODULE_SYMBOL(app_io_write),
+    DEFINE_MODULE_SYMBOL(app_io_close),
     // app/manager
     DEFINE_MODULE_SYMBOL(app_manager_start),
     DEFINE_MODULE_SYMBOL(app_manager_start_with_parameters),
     DEFINE_MODULE_SYMBOL(app_manager_start_for_result),
+    DEFINE_MODULE_SYMBOL(app_manager_start_with_streams),
     DEFINE_MODULE_SYMBOL(app_manager_stop),
     DEFINE_MODULE_SYMBOL(app_manager_get_state),
     DEFINE_MODULE_SYMBOL(app_manager_find_manifest),
@@ -53,6 +60,13 @@ static const ModuleSymbol SYMBOLS[] = {
     DEFINE_MODULE_SYMBOL(app_paths_get_assets_path),
     // app/scheduler
     DEFINE_MODULE_SYMBOL(app_scheduler_current_app_id),
+    // app/stream
+    DEFINE_MODULE_SYMBOL(app_stream_subscribe),
+    DEFINE_MODULE_SYMBOL(app_stream_unsubscribe),
+    DEFINE_MODULE_SYMBOL(app_stream_await),
+    DEFINE_MODULE_SYMBOL(app_stream_read),
+    DEFINE_MODULE_SYMBOL(app_stream_write),
+    DEFINE_MODULE_SYMBOL(app_stream_close),
     // terminator
     MODULE_SYMBOL_TERMINATOR,
 };

Modules/app-module/source/null_device.cpp

@@ -0,0 +1,45 @@
+// SPDX-License-Identifier: Apache-2.0
+#include <app/private/null_device.h>
+
+namespace {
+
+ssize_t null_read(void*, void*, size_t) {
+    return 0; // EOF
+}
+
+ssize_t null_write(void*, const void* buffer, size_t size) {
+    (void)buffer;
+    return static_cast<ssize_t>(size); // discarded, reported as fully written
+}
+
+error_t null_close(void*) {
+    return ERROR_NONE;
+}
+
+error_t null_await(void*, AppFileWait, TickType_t) {
+    return ERROR_NONE; // always ready
+}
+
+uint32_t null_poll(void*) {
+    return APP_FILE_READABLE | APP_FILE_WRITABLE;
+}
+
+const AppFileOps NULL_OPS = {
+    .read = null_read,
+    .write = null_write,
+    .close = null_close,
+    .await = null_await,
+    .poll = null_poll,
+};
+
+constexpr AppFile NULL_FILE = { .ops = &NULL_OPS, .object = nullptr };
+
+} // namespace
+
+extern "C" {
+
+const AppFile* app_null_file(void) {
+    return &NULL_FILE;
+}
+
+} // extern "C"

Modules/app-module/source/scheduler.cpp

@@ -2,9 +2,12 @@
 #include <app/instance.h>
 #include <app/loader.h>
 #include <app/private/event.h>
+#include <app/private/fd_table.h>
 #include <app/private/ledger.h>
 #include <app/private/scheduler.h>
+#include <app/private/stream_internal.h>
 #include <app/scheduler.h>
+#include <app/stream.h>
 
 #include <service/instance.h>
 #include <service/manager.h>
@@ -98,6 +101,14 @@ void set_task(AppInstanceId app_instance_id, TaskHandle_t task) {
     auto iterator = ledger.instances.find(app_instance_id);
     if (iterator != ledger.instances.end()) {
         iterator->second.task = task;
+        // Streams bound before this instance's task existed (app_manager_start_with_streams())
+        // only got producer_task filled in as NULL at subscribe time. Backfill it now.
+        AppFdTable& fd_table = iterator->second.fd_table;
+        for (auto& slot : fd_table.slots) {
+            if (slot.in_use && slot.file.ops == app_stream_ops()) {
+                static_cast<AppStream*>(slot.file.object)->producer_task = task;
+            }
+        }
     }
     mutex_unlock(&ledger.mutex);
 }
@@ -218,8 +229,14 @@ void app_task_main(void* context) {
     // Erase the ledger entry before self-deleting - see "Reap self-terminated app tasks":
     // nothing else is guaranteed to ever call app_scheduler_stop() for this instance (the
     // common case is the app just closing itself), so this can't wait for that to happen.
+    // Every non-null fd is closed here too. Stream-backed entries wake/mark closed whoever is
+    // on the other end (e.g. a parent reading this instance's stdout).
     auto& ledger = app_ledger();
     mutex_lock(&ledger.mutex);
+    auto fd_table_iterator = ledger.instances.find(app_instance_id);
+    if (fd_table_iterator != ledger.instances.end()) {
+        app_fd_table_teardown(&fd_table_iterator->second.fd_table);
+    }
     ledger.instances.erase(app_instance_id);
     mutex_unlock(&ledger.mutex);
 

Modules/app-module/source/stream.cpp

@@ -0,0 +1,313 @@
+// SPDX-License-Identifier: Apache-2.0
+#include <app/stream.h>
+
+#include <app/private/fd_table.h>
+#include <app/private/ledger.h>
+#include <app/private/stream_internal.h>
+
+#include <tactility/concurrent/mutex.h>
+#include <tactility/error.h>
+
+namespace {
+
+// Caller must hold stream->mutex.
+bool is_readable_locked(AppStream* stream) {
+    return stream->buffer.count > 0 || stream->closed;
+}
+
+// Caller must hold stream->mutex.
+bool is_writable_locked(AppStream* stream) {
+    return stream->buffer.count < stream->buffer.capacity || stream->closed;
+}
+
+// Brackets one AppFileOps call against `stream` for app_stream_unsubscribe()'s drain wait: a
+// call already blocked in app_stream_await() when unsubscribe starts only wakes, it doesn't
+// vanish, so unsubscribe must know it's still running before destructing stream->mutex.
+class StreamOperationGuard {
+public:
+    explicit StreamOperationGuard(AppStream* stream) : stream_(stream) {
+        mutex_lock(&stream_->mutex);
+        stream_->active_operations++;
+        mutex_unlock(&stream_->mutex);
+    }
+
+    ~StreamOperationGuard() {
+        mutex_lock(&stream_->mutex);
+        stream_->active_operations--;
+        mutex_unlock(&stream_->mutex);
+    }
+
+    StreamOperationGuard(const StreamOperationGuard&) = delete;
+    StreamOperationGuard& operator=(const StreamOperationGuard&) = delete;
+
+private:
+    AppStream* stream_;
+};
+
+ssize_t stream_file_read(void* object, void* buffer, size_t size) {
+    auto* stream = static_cast<AppStream*>(object);
+    StreamOperationGuard guard(stream);
+    while (true) {
+        if (app_stream_await(stream, APP_FILE_WAIT_READABLE, portMAX_DELAY) != ERROR_NONE) {
+            return -1;
+        }
+        size_t read = app_stream_read(stream, buffer, size);
+        if (read > 0) {
+            return static_cast<ssize_t>(read);
+        }
+        mutex_lock(&stream->mutex);
+        bool is_closed = stream->closed;
+        mutex_unlock(&stream->mutex);
+        if (is_closed) {
+            return 0; // EOF: readable-because-closed, and nothing left buffered
+        }
+        // Woke readable but another party drained it first. Re-await rather than assume EOF;
+        // this is a defensive fallback since AppStream expects a single reader.
+    }
+}
+
+ssize_t stream_file_write(void* object, const void* buffer, size_t size) {
+    auto* stream = static_cast<AppStream*>(object);
+    StreamOperationGuard guard(stream);
+    if (app_stream_await(stream, APP_FILE_WAIT_WRITABLE, portMAX_DELAY) != ERROR_NONE) {
+        return -1;
+    }
+    // app_stream_write() itself refuses to copy anything once closed (checked under the same
+    // lock as the copy, so a close() racing right after this await() can't slip bytes in), and
+    // reports that as 0, indistinguishable here from "woke writable, nothing to copy" without
+    // checking closed separately.
+    size_t written = app_stream_write(stream, buffer, size);
+    if (written > 0) {
+        return static_cast<ssize_t>(written);
+    }
+    mutex_lock(&stream->mutex);
+    bool is_closed = stream->closed;
+    mutex_unlock(&stream->mutex);
+    return is_closed ? -1 : 0; // broken pipe, or a spurious wake with nothing to copy
+}
+
+error_t stream_file_close(void* object) {
+    auto* stream = static_cast<AppStream*>(object);
+    StreamOperationGuard guard(stream);
+    return app_stream_close(stream);
+}
+
+error_t stream_file_await(void* object, AppFileWait wait, TickType_t timeout) {
+    auto* stream = static_cast<AppStream*>(object);
+    StreamOperationGuard guard(stream);
+    return app_stream_await(stream, wait, timeout);
+}
+
+uint32_t stream_file_poll(void* object) {
+    auto* stream = static_cast<AppStream*>(object);
+    StreamOperationGuard guard(stream);
+    mutex_lock(&stream->mutex);
+    uint32_t bits = 0;
+    if (is_readable_locked(stream)) {
+        bits |= APP_FILE_READABLE;
+    }
+    if (is_writable_locked(stream)) {
+        bits |= APP_FILE_WRITABLE;
+    }
+    mutex_unlock(&stream->mutex);
+    return bits;
+}
+
+// Fused with app_fd_table_get_and_retain()'s own lock, so a caller that gets a live AppFile back
+// is already counted here before it can be paused and raced by app_stream_unsubscribe()'s drain
+// wait (see StreamOperationGuard above for the equivalent per-call bracketing).
+void stream_file_retain(void* object) {
+    auto* stream = static_cast<AppStream*>(object);
+    mutex_lock(&stream->mutex);
+    stream->active_operations++;
+    mutex_unlock(&stream->mutex);
+}
+
+void stream_file_release(void* object) {
+    auto* stream = static_cast<AppStream*>(object);
+    mutex_lock(&stream->mutex);
+    stream->active_operations--;
+    mutex_unlock(&stream->mutex);
+}
+
+constexpr AppFileOps STREAM_OPS = {
+    .read = stream_file_read,
+    .write = stream_file_write,
+    .close = stream_file_close,
+    .await = stream_file_await,
+    .poll = stream_file_poll,
+    .retain = stream_file_retain,
+    .release = stream_file_release,
+};
+
+} // namespace
+
+extern "C" {
+
+const AppFileOps* app_stream_ops(void) {
+    return &STREAM_OPS;
+}
+
+error_t app_stream_subscribe(AppStream* stream, void* buffer, size_t buffer_capacity, TaskEventGroup* event_group, AppInstanceId producer_id, int producer_fd) {
+    if (producer_fd < 0 || producer_fd >= APP_MAX_FDS) {
+        return ERROR_OUT_OF_RANGE;
+    }
+
+    uint32_t readable_bit;
+    error_t claim_result = task_event_group_claim_bit(event_group, &readable_bit);
+    if (claim_result != ERROR_NONE) {
+        return claim_result;
+    }
+    uint32_t writable_bit;
+    claim_result = task_event_group_claim_bit(event_group, &writable_bit);
+    if (claim_result != ERROR_NONE) {
+        task_event_group_release_bit(event_group, readable_bit);
+        return claim_result;
+    }
+
+    stream->producer_id = producer_id;
+    stream->producer_fd = producer_fd;
+    stream->event_group = event_group;
+    stream->readable_bit = readable_bit;
+    stream->writable_bit = writable_bit;
+    stream->buffer.data = static_cast<uint8_t*>(buffer);
+    stream->buffer.capacity = buffer_capacity;
+    stream->buffer.read_pos = 0;
+    stream->buffer.write_pos = 0;
+    stream->buffer.count = 0;
+    stream->closed = false;
+    stream->active_operations = 0;
+    mutex_construct(&stream->mutex);
+
+    // Held across the fd-table lookup and bind so this can't race app_fd_table_teardown():
+    // both teardown call sites (scheduler.cpp, manager.cpp) hold this same ledger mutex for
+    // their whole teardown call, and app_fd_table_bind()/close()/get() check for it (see
+    // fd_table.h's AppFdTable::shutting_down).
+    auto& ledger = app_ledger();
+    mutex_lock(&ledger.mutex);
+    auto iterator = ledger.instances.find(producer_id);
+    if (iterator == ledger.instances.end()) {
+        mutex_unlock(&ledger.mutex);
+        mutex_destruct(&stream->mutex);
+        task_event_group_release_bit(event_group, readable_bit);
+        task_event_group_release_bit(event_group, writable_bit);
+        return ERROR_NOT_FOUND;
+    }
+    stream->producer_task = iterator->second.task;
+    error_t bind_result = app_fd_table_bind(&iterator->second.fd_table, producer_fd, &STREAM_OPS, stream);
+    mutex_unlock(&ledger.mutex);
+
+    if (bind_result != ERROR_NONE) {
+        mutex_destruct(&stream->mutex);
+        task_event_group_release_bit(event_group, readable_bit);
+        task_event_group_release_bit(event_group, writable_bit);
+    }
+    return bind_result;
+}
+
+error_t app_stream_unsubscribe(AppStream* stream) {
+    // Held across the fd-table lookup, get, and close. See app_stream_subscribe()'s own
+    // comment on why this must stay exclusive with app_fd_table_teardown().
+    auto& ledger = app_ledger();
+    mutex_lock(&ledger.mutex);
+    auto iterator = ledger.instances.find(stream->producer_id);
+    if (iterator != ledger.instances.end()) {
+        AppFdTable* table = &iterator->second.fd_table;
+        AppFile current {};
+        if (app_fd_table_get(table, stream->producer_fd, &current) && current.object == stream) {
+            app_fd_table_close(table, stream->producer_fd); // -> stream_file_close() -> app_stream_close()
+        }
+    }
+    mutex_unlock(&ledger.mutex);
+
+    app_stream_close(stream); // idempotent; wakes anyone already blocked in app_stream_await()
+
+    // Waking a blocked call doesn't mean it has finished. It still has to get scheduled, notice
+    // it's closed, and return. Wait for that before destructing mutex/bits below, mirroring
+    // scheduler.cpp's reap_self()/reaper_task_main() waiting out a task before freeing its stack.
+    while (true) {
+        mutex_lock(&stream->mutex);
+        int active_operations = stream->active_operations;
+        mutex_unlock(&stream->mutex);
+        if (active_operations == 0) {
+            break;
+        }
+        taskYIELD();
+    }
+
+    task_event_group_release_bit(stream->event_group, stream->readable_bit);
+    task_event_group_release_bit(stream->event_group, stream->writable_bit);
+    mutex_destruct(&stream->mutex);
+    return ERROR_NONE;
+}
+
+error_t app_stream_await(AppStream* stream, AppFileWait wait, TickType_t timeout) {
+    mutex_lock(&stream->mutex);
+    bool ready = (wait == APP_FILE_WAIT_READABLE) ? is_readable_locked(stream) : is_writable_locked(stream);
+    mutex_unlock(&stream->mutex);
+    if (ready) {
+        return ERROR_NONE;
+    }
+
+    uint32_t bit = (wait == APP_FILE_WAIT_READABLE) ? stream->readable_bit : stream->writable_bit;
+    return task_event_group_wait(stream->event_group, bit, /*await_all=*/false, nullptr, timeout);
+}
+
+size_t app_stream_read(AppStream* stream, void* buffer, size_t buffer_size) {
+    mutex_lock(&stream->mutex);
+    AppStreamBuffer& ring = stream->buffer;
+    size_t to_copy = buffer_size < ring.count ? buffer_size : ring.count;
+    for (size_t i = 0; i < to_copy; i++) {
+        static_cast<uint8_t*>(buffer)[i] = ring.data[(ring.read_pos + i) % ring.capacity];
+    }
+    if (ring.capacity > 0) {
+        ring.read_pos = (ring.read_pos + to_copy) % ring.capacity;
+    }
+    ring.count -= to_copy;
+    TaskEventGroup* event_group = stream->event_group;
+    uint32_t writable_bit = stream->writable_bit;
+    mutex_unlock(&stream->mutex);
+    if (to_copy > 0) {
+        task_event_group_signal(event_group, writable_bit); // space freed up
+    }
+    return to_copy;
+}
+
+size_t app_stream_write(AppStream* stream, const void* buffer, size_t buffer_size) {
+    mutex_lock(&stream->mutex);
+    if (stream->closed) {
+        mutex_unlock(&stream->mutex);
+        return 0;
+    }
+    AppStreamBuffer& ring = stream->buffer;
+    size_t available = ring.capacity - ring.count;
+    size_t to_copy = buffer_size < available ? buffer_size : available;
+    for (size_t i = 0; i < to_copy; i++) {
+        ring.data[(ring.write_pos + i) % ring.capacity] = static_cast<const uint8_t*>(buffer)[i];
+    }
+    if (ring.capacity > 0) {
+        ring.write_pos = (ring.write_pos + to_copy) % ring.capacity;
+    }
+    ring.count += to_copy;
+    TaskEventGroup* event_group = stream->event_group;
+    uint32_t readable_bit = stream->readable_bit;
+    mutex_unlock(&stream->mutex);
+    if (to_copy > 0) {
+        task_event_group_signal(event_group, readable_bit); // data became available
+    }
+    return to_copy;
+}
+
+error_t app_stream_close(AppStream* stream) {
+    mutex_lock(&stream->mutex);
+    stream->closed = true;
+    TaskEventGroup* event_group = stream->event_group;
+    uint32_t readable_bit = stream->readable_bit;
+    uint32_t writable_bit = stream->writable_bit;
+    mutex_unlock(&stream->mutex);
+    task_event_group_signal(event_group, readable_bit);
+    task_event_group_signal(event_group, writable_bit);
+    return ERROR_NONE;
+}
+
+} // extern "C"