Step 321: TaskQueue — Priority Queue with Dependencies (12/12 tests)
Ordered queue with priority scheduling and dependency resolution. Items blocked by unsatisfied dependencies stay pending; completing an item cascades to unblock dependents. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bill
@@ -1921,4 +1921,9 @@ add_executable(step320_test tests/step320_test.cpp)
|
||||
target_include_directories(step320_test PRIVATE src)
|
||||
target_link_libraries(step320_test PRIVATE nlohmann_json::nlohmann_json)
|
||||
|
||||
# Step 321: TaskQueue — Priority Queue with Dependencies
|
||||
add_executable(step321_test tests/step321_test.cpp)
|
||||
target_include_directories(step321_test PRIVATE src)
|
||||
target_link_libraries(step321_test PRIVATE nlohmann_json::nlohmann_json)
|
||||
|
||||
# Step 12: Dear ImGui shell scaffolding created (main.cpp exists but not built due to dependencies)
|
||||
|
||||
200
editor/src/TaskQueue.h
Normal file
200
editor/src/TaskQueue.h
Normal file
@@ -0,0 +1,200 @@
|
||||
#pragma once
|
||||
// Step 321: TaskQueue — Priority Queue with Dependencies
|
||||
//
|
||||
// Ordered queue that respects both priority levels and dependency chains.
|
||||
// Items whose dependencies are unsatisfied stay blocked; ready items are
|
||||
// ordered by priority then creation time.
|
||||
|
||||
#include "WorkItem.h"
|
||||
#include <vector>
|
||||
#include <optional>
|
||||
#include <algorithm>
|
||||
#include <map>
|
||||
|
||||
class TaskQueue {
|
||||
public:
|
||||
// Add a work item to the queue
|
||||
void enqueue(WorkItem item) {
|
||||
// Set status based on whether dependencies exist
|
||||
if (item.dependencies.empty()) {
|
||||
if (item.status == WI_PENDING) {
|
||||
transitionWorkItem(item, WI_READY);
|
||||
}
|
||||
}
|
||||
// else stays pending until dependencies complete
|
||||
items_.push_back(std::move(item));
|
||||
}
|
||||
|
||||
// Returns items whose dependencies are all complete, ordered by priority then createdAt
|
||||
std::vector<WorkItem> getReady() const {
|
||||
std::vector<WorkItem> ready;
|
||||
for (const auto& item : items_) {
|
||||
if (item.status == WI_READY) {
|
||||
ready.push_back(item);
|
||||
}
|
||||
}
|
||||
sortByPriority(ready);
|
||||
return ready;
|
||||
}
|
||||
|
||||
// Next ready item without removing
|
||||
std::optional<WorkItem> peek() const {
|
||||
auto ready = getReady();
|
||||
if (ready.empty()) return std::nullopt;
|
||||
return ready.front();
|
||||
}
|
||||
|
||||
// Remove and return next ready item, transition to assigned
|
||||
std::optional<WorkItem> dequeue() {
|
||||
auto ready = getReady();
|
||||
if (ready.empty()) return std::nullopt;
|
||||
|
||||
std::string id = ready.front().id;
|
||||
for (auto& item : items_) {
|
||||
if (item.id == id) {
|
||||
transitionWorkItem(item, WI_ASSIGNED);
|
||||
return item;
|
||||
}
|
||||
}
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
// Mark complete, trigger dependency re-evaluation
|
||||
bool complete(const std::string& itemId) {
|
||||
auto* item = findItem(itemId);
|
||||
if (!item) return false;
|
||||
|
||||
// Transition through necessary states to reach complete
|
||||
if (item->status == WI_IN_PROGRESS || item->status == WI_REVIEW) {
|
||||
transitionWorkItem(*item, WI_COMPLETE);
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
|
||||
// Re-evaluate blocked items
|
||||
resolveDependencies();
|
||||
return true;
|
||||
}
|
||||
|
||||
// Mark rejected, re-enqueue as ready with feedback
|
||||
bool reject(const std::string& itemId, const std::string& feedback) {
|
||||
auto* item = findItem(itemId);
|
||||
if (!item) return false;
|
||||
|
||||
if (item->status != WI_REVIEW) return false;
|
||||
|
||||
transitionWorkItem(*item, WI_REJECTED);
|
||||
item->result.reasoning = feedback;
|
||||
|
||||
// Re-enqueue: rejected → ready
|
||||
transitionWorkItem(*item, WI_READY);
|
||||
return true;
|
||||
}
|
||||
|
||||
// Items waiting on dependencies
|
||||
std::vector<WorkItem> getBlocked() const {
|
||||
std::vector<WorkItem> blocked;
|
||||
for (const auto& item : items_) {
|
||||
if (item.status == WI_PENDING && !item.dependencies.empty()) {
|
||||
blocked.push_back(item);
|
||||
}
|
||||
}
|
||||
return blocked;
|
||||
}
|
||||
|
||||
// Filter by lifecycle status
|
||||
std::vector<WorkItem> getByStatus(const std::string& status) const {
|
||||
std::vector<WorkItem> result;
|
||||
for (const auto& item : items_) {
|
||||
if (item.status == status) {
|
||||
result.push_back(item);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
int size() const { return static_cast<int>(items_.size()); }
|
||||
|
||||
int readyCount() const {
|
||||
int count = 0;
|
||||
for (const auto& item : items_) {
|
||||
if (item.status == WI_READY) ++count;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
int blockedCount() const {
|
||||
int count = 0;
|
||||
for (const auto& item : items_) {
|
||||
if (item.status == WI_PENDING && !item.dependencies.empty()) ++count;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
// Lookup by ID
|
||||
std::optional<WorkItem> getItem(const std::string& itemId) const {
|
||||
for (const auto& item : items_) {
|
||||
if (item.id == itemId) return item;
|
||||
}
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
// Replace item (for external status updates)
|
||||
bool updateItem(const std::string& itemId, const WorkItem& updated) {
|
||||
for (auto& item : items_) {
|
||||
if (item.id == itemId) {
|
||||
item = updated;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private:
|
||||
std::vector<WorkItem> items_;
|
||||
|
||||
WorkItem* findItem(const std::string& id) {
|
||||
for (auto& item : items_) {
|
||||
if (item.id == id) return &item;
|
||||
}
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
// Check if a given item ID is complete
|
||||
bool isComplete(const std::string& id) const {
|
||||
for (const auto& item : items_) {
|
||||
if (item.id == id) return item.status == WI_COMPLETE;
|
||||
}
|
||||
// Dependency not in queue — treat as satisfied
|
||||
return true;
|
||||
}
|
||||
|
||||
// Scan blocked items and promote to ready if all deps complete
|
||||
void resolveDependencies() {
|
||||
for (auto& item : items_) {
|
||||
if (item.status != WI_PENDING) continue;
|
||||
if (item.dependencies.empty()) continue;
|
||||
|
||||
bool allComplete = true;
|
||||
for (const auto& dep : item.dependencies) {
|
||||
if (!isComplete(dep)) {
|
||||
allComplete = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (allComplete) {
|
||||
transitionWorkItem(item, WI_READY);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Sort by priority (lower int = higher priority), then by createdAt
|
||||
static void sortByPriority(std::vector<WorkItem>& items) {
|
||||
std::sort(items.begin(), items.end(), [](const WorkItem& a, const WorkItem& b) {
|
||||
int pa = priorityToInt(a.priority);
|
||||
int pb = priorityToInt(b.priority);
|
||||
if (pa != pb) return pa < pb;
|
||||
return a.createdAt < b.createdAt;
|
||||
});
|
||||
}
|
||||
};
|
||||
301
editor/tests/step321_test.cpp
Normal file
301
editor/tests/step321_test.cpp
Normal file
@@ -0,0 +1,301 @@
|
||||
// Step 321: TaskQueue — Priority Queue with Dependencies (12 tests)
|
||||
// Tests enqueue/dequeue ordering, dependency blocking, reject/re-enqueue,
|
||||
// size tracking, completion cascading, mixed priorities with dependencies.
|
||||
|
||||
#include "TaskQueue.h"
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <chrono>
|
||||
|
||||
static int passed = 0, failed = 0;
|
||||
#define TEST(name) { std::cout << " " << #name << "... "; }
|
||||
#define PASS() { std::cout << "PASS\n"; ++passed; }
|
||||
#define FAIL(msg) { std::cout << "FAIL: " << msg << "\n"; ++failed; }
|
||||
#define CHECK(cond, msg) if (!(cond)) { FAIL(msg); return; } else {}
|
||||
|
||||
// Helper: create a WorkItem with given id, priority, and optional dependencies
|
||||
static WorkItem makeItem(const std::string& id, const std::string& priority,
|
||||
const std::vector<std::string>& deps = {}) {
|
||||
WorkItem wi;
|
||||
wi.id = id;
|
||||
wi.nodeId = "node_" + id;
|
||||
wi.nodeName = "func_" + id;
|
||||
wi.nodeType = "Function";
|
||||
wi.bufferId = "buf";
|
||||
wi.contextWidth = "local";
|
||||
wi.workerType = "llm";
|
||||
wi.priority = priority;
|
||||
wi.dependencies = deps;
|
||||
wi.status = WI_PENDING;
|
||||
wi.createdAt = workItemTimestamp();
|
||||
return wi;
|
||||
}
|
||||
|
||||
// 1. Enqueue/dequeue basic ordering
|
||||
void test_enqueue_dequeue() {
|
||||
TEST(enqueue_dequeue);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("a", "medium"));
|
||||
q.enqueue(makeItem("b", "medium"));
|
||||
|
||||
CHECK(q.size() == 2, "size=2");
|
||||
auto item = q.dequeue();
|
||||
CHECK(item.has_value(), "dequeue returns item");
|
||||
CHECK(item->id == "a", "FIFO for same priority: got " + item->id);
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 2. Priority ordering — critical before low
|
||||
void test_priority_ordering() {
|
||||
TEST(priority_ordering);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("low1", "low"));
|
||||
q.enqueue(makeItem("crit1", "critical"));
|
||||
q.enqueue(makeItem("med1", "medium"));
|
||||
q.enqueue(makeItem("high1", "high"));
|
||||
|
||||
auto ready = q.getReady();
|
||||
CHECK(ready.size() == 4, "4 ready");
|
||||
CHECK(ready[0].id == "crit1", "first=critical");
|
||||
CHECK(ready[1].id == "high1", "second=high");
|
||||
CHECK(ready[2].id == "med1", "third=medium");
|
||||
CHECK(ready[3].id == "low1", "fourth=low");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 3. Dependency blocking — B blocked by A, A completes → B becomes ready
|
||||
void test_dependency_blocking() {
|
||||
TEST(dependency_blocking);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("A", "medium"));
|
||||
q.enqueue(makeItem("B", "medium", {"A"}));
|
||||
|
||||
CHECK(q.readyCount() == 1, "1 ready (A)");
|
||||
CHECK(q.blockedCount() == 1, "1 blocked (B)");
|
||||
|
||||
// Dequeue A, progress it through to complete
|
||||
auto a = q.dequeue();
|
||||
CHECK(a.has_value(), "dequeued A");
|
||||
// Manually transition: assigned → in-progress → complete
|
||||
auto aItem = q.getItem("A");
|
||||
CHECK(aItem.has_value(), "A still in queue");
|
||||
WorkItem updated = *aItem;
|
||||
transitionWorkItem(updated, WI_IN_PROGRESS);
|
||||
q.updateItem("A", updated);
|
||||
q.complete("A");
|
||||
|
||||
CHECK(q.readyCount() == 1, "B now ready");
|
||||
CHECK(q.blockedCount() == 0, "0 blocked");
|
||||
auto b = q.peek();
|
||||
CHECK(b.has_value() && b->id == "B", "B is next");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 4. Reject and re-enqueue
|
||||
void test_reject_reenqueue() {
|
||||
TEST(reject_reenqueue);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("X", "high"));
|
||||
|
||||
auto x = q.dequeue();
|
||||
CHECK(x.has_value(), "dequeued X");
|
||||
|
||||
// Move to in-progress → review
|
||||
WorkItem updated = *q.getItem("X");
|
||||
transitionWorkItem(updated, WI_IN_PROGRESS);
|
||||
q.updateItem("X", updated);
|
||||
updated = *q.getItem("X");
|
||||
transitionWorkItem(updated, WI_REVIEW);
|
||||
q.updateItem("X", updated);
|
||||
|
||||
bool rejected = q.reject("X", "needs better error handling");
|
||||
CHECK(rejected, "reject succeeded");
|
||||
|
||||
auto xNow = q.getItem("X");
|
||||
CHECK(xNow.has_value(), "X still exists");
|
||||
CHECK(xNow->status == WI_READY, "X is ready again");
|
||||
CHECK(xNow->result.reasoning == "needs better error handling", "feedback preserved");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 5. getReady returns only unblocked items
|
||||
void test_get_ready_filtering() {
|
||||
TEST(get_ready_filtering);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("r1", "medium"));
|
||||
q.enqueue(makeItem("r2", "medium"));
|
||||
q.enqueue(makeItem("b1", "medium", {"r1"}));
|
||||
q.enqueue(makeItem("b2", "medium", {"r2"}));
|
||||
|
||||
auto ready = q.getReady();
|
||||
CHECK(ready.size() == 2, "2 ready, got " + std::to_string(ready.size()));
|
||||
// blocked items should not appear
|
||||
for (const auto& r : ready) {
|
||||
CHECK(r.id == "r1" || r.id == "r2", "only r1/r2 in ready");
|
||||
}
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 6. Empty queue behavior
|
||||
void test_empty_queue() {
|
||||
TEST(empty_queue);
|
||||
TaskQueue q;
|
||||
CHECK(q.size() == 0, "empty size");
|
||||
CHECK(q.readyCount() == 0, "empty readyCount");
|
||||
CHECK(q.blockedCount() == 0, "empty blockedCount");
|
||||
CHECK(!q.peek().has_value(), "peek returns nullopt");
|
||||
CHECK(!q.dequeue().has_value(), "dequeue returns nullopt");
|
||||
CHECK(!q.getItem("nonexistent").has_value(), "getItem returns nullopt");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 7. getByStatus filtering
|
||||
void test_get_by_status() {
|
||||
TEST(get_by_status);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("s1", "medium"));
|
||||
q.enqueue(makeItem("s2", "medium"));
|
||||
q.enqueue(makeItem("s3", "medium", {"s1"}));
|
||||
|
||||
auto ready = q.getByStatus(WI_READY);
|
||||
CHECK(ready.size() == 2, "2 ready");
|
||||
auto pending = q.getByStatus(WI_PENDING);
|
||||
CHECK(pending.size() == 1, "1 pending");
|
||||
|
||||
q.dequeue(); // assign s1
|
||||
auto assigned = q.getByStatus(WI_ASSIGNED);
|
||||
CHECK(assigned.size() == 1, "1 assigned");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 8. Size tracking after operations
|
||||
void test_size_tracking() {
|
||||
TEST(size_tracking);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("t1", "low"));
|
||||
q.enqueue(makeItem("t2", "high"));
|
||||
q.enqueue(makeItem("t3", "medium", {"t1"}));
|
||||
|
||||
CHECK(q.size() == 3, "size=3");
|
||||
CHECK(q.readyCount() == 2, "readyCount=2");
|
||||
CHECK(q.blockedCount() == 1, "blockedCount=1");
|
||||
|
||||
q.dequeue(); // dequeue t2 (highest priority)
|
||||
CHECK(q.readyCount() == 1, "readyCount=1 after dequeue");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 9. Completion cascading (A → B → C chain)
|
||||
void test_completion_cascade() {
|
||||
TEST(completion_cascade);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("c1", "high"));
|
||||
q.enqueue(makeItem("c2", "high", {"c1"}));
|
||||
q.enqueue(makeItem("c3", "high", {"c2"}));
|
||||
|
||||
CHECK(q.readyCount() == 1, "only c1 ready");
|
||||
CHECK(q.blockedCount() == 2, "c2, c3 blocked");
|
||||
|
||||
// Complete c1
|
||||
q.dequeue(); // assign c1
|
||||
WorkItem u = *q.getItem("c1");
|
||||
transitionWorkItem(u, WI_IN_PROGRESS);
|
||||
q.updateItem("c1", u);
|
||||
q.complete("c1");
|
||||
|
||||
CHECK(q.readyCount() == 1, "c2 now ready");
|
||||
CHECK(q.blockedCount() == 1, "c3 still blocked");
|
||||
|
||||
// Complete c2
|
||||
q.dequeue(); // assign c2
|
||||
u = *q.getItem("c2");
|
||||
transitionWorkItem(u, WI_IN_PROGRESS);
|
||||
q.updateItem("c2", u);
|
||||
q.complete("c2");
|
||||
|
||||
CHECK(q.readyCount() == 1, "c3 now ready");
|
||||
CHECK(q.blockedCount() == 0, "none blocked");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 10. Mixed priorities with dependencies
|
||||
void test_mixed_priorities_deps() {
|
||||
TEST(mixed_priorities_deps);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("lo", "low"));
|
||||
q.enqueue(makeItem("hi", "critical", {"lo"}));
|
||||
|
||||
// hi is critical but blocked, lo is low but ready
|
||||
auto ready = q.getReady();
|
||||
CHECK(ready.size() == 1, "1 ready");
|
||||
CHECK(ready[0].id == "lo", "lo is ready despite lower priority");
|
||||
|
||||
// Complete lo → hi becomes ready
|
||||
q.dequeue();
|
||||
WorkItem u = *q.getItem("lo");
|
||||
transitionWorkItem(u, WI_IN_PROGRESS);
|
||||
q.updateItem("lo", u);
|
||||
q.complete("lo");
|
||||
|
||||
ready = q.getReady();
|
||||
CHECK(ready.size() == 1, "hi now ready");
|
||||
CHECK(ready[0].id == "hi", "hi unblocked");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 11. Peek does not remove
|
||||
void test_peek_no_remove() {
|
||||
TEST(peek_no_remove);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("p1", "medium"));
|
||||
|
||||
auto first = q.peek();
|
||||
CHECK(first.has_value(), "peek returns value");
|
||||
auto second = q.peek();
|
||||
CHECK(second.has_value(), "peek still returns value");
|
||||
CHECK(first->id == second->id, "same item");
|
||||
CHECK(q.readyCount() == 1, "still 1 ready");
|
||||
PASS();
|
||||
}
|
||||
|
||||
// 12. updateItem replaces item data
|
||||
void test_update_item() {
|
||||
TEST(update_item);
|
||||
TaskQueue q;
|
||||
q.enqueue(makeItem("u1", "medium"));
|
||||
|
||||
auto item = *q.getItem("u1");
|
||||
item.assignee = "worker-42";
|
||||
item.workerType = "deterministic";
|
||||
q.updateItem("u1", item);
|
||||
|
||||
auto updated = q.getItem("u1");
|
||||
CHECK(updated.has_value(), "item exists");
|
||||
CHECK(updated->assignee == "worker-42", "assignee updated");
|
||||
CHECK(updated->workerType == "deterministic", "workerType updated");
|
||||
PASS();
|
||||
}
|
||||
|
||||
int main() {
|
||||
std::cout << "=== Step 321: TaskQueue Priority Queue with Dependencies ===\n";
|
||||
try {
|
||||
test_enqueue_dequeue();
|
||||
test_priority_ordering();
|
||||
test_dependency_blocking();
|
||||
test_reject_reenqueue();
|
||||
test_get_ready_filtering();
|
||||
test_empty_queue();
|
||||
test_get_by_status();
|
||||
test_size_tracking();
|
||||
test_completion_cascade();
|
||||
test_mixed_priorities_deps();
|
||||
test_peek_no_remove();
|
||||
test_update_item();
|
||||
} catch (const std::exception& e) {
|
||||
std::cout << "EXCEPTION: " << e.what() << "\n";
|
||||
++failed;
|
||||
}
|
||||
std::cout << "\nResults: " << passed << "/" << (passed + failed) << " passed\n";
|
||||
return failed > 0 ? 1 : 0;
|
||||
}
|
||||
19
progress.md
19
progress.md
@@ -1226,6 +1226,25 @@ work that flows through the routing engine.
|
||||
|
||||
**State machine:** pending→ready→assigned→in-progress→(review→complete|rejected OR complete); rejected→ready
|
||||
|
||||
### Step 321: TaskQueue — Priority Queue with Dependencies
|
||||
**Status:** PASS (12/12 tests)
|
||||
|
||||
Ordered queue that respects both priority levels and dependency chains.
|
||||
Items whose dependencies are unsatisfied stay blocked; ready items are ordered
|
||||
by priority (critical first) then creation time.
|
||||
|
||||
**Files created:**
|
||||
- `editor/src/TaskQueue.h` — TaskQueue class with enqueue, dequeue, peek,
|
||||
complete, reject, getReady, getBlocked, getByStatus, getItem, updateItem;
|
||||
dependency resolution on completion, priority-based sorting
|
||||
- `editor/tests/step321_test.cpp` — 12 tests: enqueue/dequeue ordering,
|
||||
priority ordering, dependency blocking, reject/re-enqueue, getReady
|
||||
filtering, empty queue, getByStatus, size tracking, completion cascading
|
||||
(A→B→C chain), mixed priorities with deps, peek no-remove, updateItem
|
||||
|
||||
**Files modified:**
|
||||
- `editor/CMakeLists.txt` — step321_test target
|
||||
|
||||
---
|
||||
|
||||
# Roadmap Planning — Sprints 12-25+
|
||||
|
||||
Reference in New Issue
Block a user