Batched Constellation Editing

# In safe assignment lock algorithm
while system is running:
    foreach event e ∈ E do
        if e is TASK_COMPLETED or TASK_FAILED then
            async enqueue(e)  # ← Queue instead of immediate processing
        end
    end

    acquire(assign_lock)

    # Process ALL queued events in one batch
    while queue not empty do
        e ← dequeue()
        Δ ← invoke(ConstellationAgent, edit(C, e))
        C ← apply(C, Δ)
    end

    validate(C)  # ← Single validation for entire batch
    publish(CONSTELLATION_MODIFIED, all_task_ids)
    C ← synchronize(C, T_C)

    release(assign_lock)

async def wait_for_pending_modifications(
    self, timeout: Optional[float] = None
) -> bool:
    """Wait for all pending modifications to complete."""

    if not self._pending_modifications:
        return True

    try:
        while self._pending_modifications:
            # Get current pending tasks (snapshot)
            pending_tasks = list(self._pending_modifications.keys())
            pending_futures = list(self._pending_modifications.values())

            self.logger.info(
                f"⏳ Waiting for {len(pending_tasks)} pending modification(s): "
                f"{pending_tasks}"
            )

            # Wait for ALL current pending modifications (batching)
            await asyncio.wait_for(
                asyncio.gather(*pending_futures, return_exceptions=True),
                timeout=remaining_timeout,
            )

            # Check if new modifications were added during wait
            if not self._pending_modifications:
                break

            # Small delay to allow new registrations to settle
            await asyncio.sleep(0.01)

        self.logger.info("✅ All pending modifications completed")
        return True

    except asyncio.TimeoutError:
        ...

async def process_editing(
    self,
    context: Context = None,
    task_ids: Optional[List[str]] = None,  # ← Multiple task IDs
    before_constellation: Optional[TaskConstellation] = None,
) -> TaskConstellation:
    """Process task completion events and update constellation."""

    task_ids = task_ids or []

    # Agent can see multiple completed tasks at once
    self.logger.debug(
        f"Tasks {task_ids} marked as completed, processing modifications..."
    )

    # Potentially make decisions based on multiple task outcomes
    # e.g., "Task A and B both succeeded, skip Task C"
    after_constellation = await self._create_and_process(context)

    # Publish single CONSTELLATION_MODIFIED event for entire batch
    await self._publish_constellation_modified_event(
        before_constellation,
        after_constellation,
        task_ids,  # ← All modified tasks
        self._create_timing_info(start_time, end_time, duration),
    )

    return after_constellation

t=100ms: Task A completes
    → Synchronizer registers pending modification for A
    → Task A's event added to queue

t=150ms: Task B completes (during A's queueing)
    → Synchronizer registers pending modification for B
    → Task B's event added to queue

t=200ms: Task C completes
    → Synchronizer registers pending modification for C
    → Task C's event added to queue

t=205ms: Orchestrator reaches synchronization point
    → Calls wait_for_pending_modifications()
    → Sees pending: [A, B, C]
    → Waits for all three futures

t=210ms: Agent starts processing (lock acquired)
    → Receives task_ids = ['A', 'B', 'C']
    → Makes unified editing decision

t=350ms: Agent completes editing
    → Publishes CONSTELLATION_MODIFIED with on_task_id = ['A', 'B', 'C']

t=355ms: Synchronizer receives event
    → Completes futures for A, B, C
    → wait_for_pending_modifications() returns

t=360ms: Orchestrator merges states and continues
    → Single validation
    → Single synchronization
    → Resume scheduling

# In wait_for_pending_modifications()
while self._pending_modifications:
    # Snapshot current pending tasks
    pending_tasks = list(self._pending_modifications.keys())

    # Wait for all of them
    await asyncio.gather(*pending_futures)

    # Check for new arrivals during processing
    if not self._pending_modifications:
        break

    # If new tasks arrived, include them in next iteration

acquire(assign_lock)

# Apply all modifications together
foreach event in batch:
    Δ ← invoke(ConstellationAgent, edit(C, event))
    C ← apply(C, Δ)
end

validate(C)  # ← Validates combined result
publish(CONSTELLATION_MODIFIED, batch_task_ids)

release(assign_lock)

from galaxy.session.observers.constellation_sync_observer import (
    ConstellationModificationSynchronizer
)

# Create and attach synchronizer
synchronizer = ConstellationModificationSynchronizer(orchestrator)
orchestrator.set_modification_synchronizer(synchronizer)

# Subscribe to events
event_bus.subscribe(synchronizer)

# Batching happens automatically
results = await orchestrator.orchestrate_constellation(constellation)

# After orchestration
stats = synchronizer.get_statistics()

print(f"Total modifications: {stats['total_modifications']}")
print(f"Completed: {stats['completed_modifications']}")

# Infer average batch size
avg_batch_size = stats['total_modifications'] / number_of_edit_cycles
print(f"Average batch size: {avg_batch_size:.2f}")

# Increase timeout for complex reasoning
synchronizer.set_modification_timeout(1800.0)  # 30 minutes

# Or decrease for simple tasks
synchronizer.set_modification_timeout(120.0)  # 2 minutes

# Good: Tasks with similar durations
Task A: 5 seconds
Task B: 6 seconds  # ← Likely completes near Task A
Task C: 5 seconds  # ← Likely completes near Task A

# Bad: Widely varying durations
Task X: 1 second
Task Y: 30 seconds  # ← Won't batch with X
Task Z: 2 seconds   # ← Won't batch with Y

class BatchMetricsObserver(IEventObserver):
    def __init__(self):
        self.batch_sizes = []

    async def on_event(self, event: Event):
        if event.event_type == EventType.CONSTELLATION_MODIFIED:
            task_ids = event.data.get("on_task_id", [])
            batch_size = len(task_ids)
            self.batch_sizes.append(batch_size)

            if batch_size > 1:
                print(f"✓ Batched {batch_size} modifications")