vik 0.1.2

Vik is an issue-driven coding workflow automation tool.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

//! Typed orchestrator event channel.
//!
//! One bounded mpsc channel carries intake signals into the orchestrator
//! main loop. Stage/session signals stay inside
//! [`super::session_manager::StageSessionManager`].

use tokio::sync::mpsc;

use crate::context::Issue;

/// Bounded so a slow main loop applies backpressure to producers. 256 is
/// large enough to swallow a normal intake burst without forcing intake
/// to await mid-cycle.
const EVENT_BUFFER: usize = 256;

#[derive(Clone)]
pub(super) struct EventProducer {
  sender: mpsc::Sender<OrchestratorEvent>,
}

impl EventProducer {
  pub(super) async fn intake_issue(&self, issue: Issue) {
    self.send(OrchestratorEvent::Intake(IntakeEvent::Issue(issue))).await;
  }

  pub(super) async fn intake_failed(&self, error: impl ToString) {
    self
      .send(OrchestratorEvent::Intake(IntakeEvent::Failed(error.to_string())))
      .await;
  }

  pub(super) async fn intake_stopped(&self) {
    self.send(OrchestratorEvent::Intake(IntakeEvent::Stopped)).await;
  }

  async fn send(&self, event: OrchestratorEvent) {
    if self.sender.send(event).await.is_err() {
      tracing::debug!("orchestrator event receiver dropped");
    }
  }
}

pub(super) struct EventConsumer {
  receiver: mpsc::Receiver<OrchestratorEvent>,
}

impl EventConsumer {
  pub(super) async fn recv(&mut self) -> Option<OrchestratorEvent> {
    self.receiver.recv().await
  }
}

pub(super) fn event_channel() -> (EventProducer, EventConsumer) {
  let (sender, receiver) = mpsc::channel(EVENT_BUFFER);
  (EventProducer { sender }, EventConsumer { receiver })
}

// TODO: FIXME
#[allow(clippy::large_enum_variant)]
pub(super) enum OrchestratorEvent {
  Intake(IntakeEvent),
}

pub(super) enum IntakeEvent {
  Issue(Issue),
  /// Recoverable error during one cycle — the loop keeps going.
  Failed(String),
  /// Natural end (max iterations or cooperative shutdown). Triggers the
  /// main loop's drain check.
  Stopped,
}

#[cfg(test)]
mod tests {
  use super::*;

  #[tokio::test]
  async fn producer_delivers_intake_events_in_order() {
    let (producer, mut consumer) = event_channel();

    producer.intake_failed("pull failed").await;
    producer.intake_stopped().await;

    match consumer.recv().await.expect("first event") {
      OrchestratorEvent::Intake(IntakeEvent::Failed(error)) => assert_eq!(error, "pull failed"),
      _ => panic!("expected intake failure"),
    }

    match consumer.recv().await.expect("second event") {
      OrchestratorEvent::Intake(IntakeEvent::Stopped) => {},
      _ => panic!("expected intake stopped"),
    }
  }
}