qq-cli 0.1.0

A modular Rust QQ task interface for coding agents
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

//! Event subscription application entry; cancellation and output belong to the caller.
use crate::cli::args::{Cli, Command, EventCommand};
use crate::core::error::{AppError, Result};
use crate::messaging::session::Session;
use crate::storage::{
    config::Settings,
    credentials::{Credentials, SystemCredentials},
    references::References,
    state::{Store, select},
};
use std::time::{Duration, Instant};

/// Streaming entry point; caller owns cancellation and output process lifetime.
pub fn execute_watch(
    cli: &Cli,
    sink: &mut impl crate::messaging::watch::Sink,
    cancel: &std::sync::atomic::AtomicBool,
) -> Result<u8> {
    cli.validate()?;
    let Command::Event {
        command:
            EventCommand::Watch {
                conversation,
                limit,
                deadline,
                resume,
            },
    } = &cli.command
    else {
        return Err(AppError::invalid());
    };
    let deadline = Instant::now() + Duration::from_secs(u64::from(deadline.timeout));
    let settings = Settings::load(cli.config.as_deref())?;
    let store = Store::new(settings.state_dir);
    let registry = store.read()?;
    let instance = select(&registry, cli.instance.as_deref())?;
    crate::messaging::session::require_api_ready(&instance)?;
    if !settings.config.permissions.events || !settings.config.permissions.read {
        return Err(AppError::new(
            "PERMISSION_DENIED",
            "The local policy does not permit event observation.",
            4,
        ));
    }
    let endpoint = instance.ws_endpoint.as_deref().ok_or_else(|| {
        AppError::new(
            "EVENTS_UNAVAILABLE",
            "Selected instance has no registered forward event endpoint.",
            4,
        )
    })?;
    crate::transport::websocket::validate_endpoint(endpoint)?;
    let installation = registry
        .installations
        .iter()
        .find(|item| item.id == instance.installation_id)
        .ok_or_else(|| {
            AppError::new(
                "INSTALLATION_NOT_FOUND",
                "Instance installation is not registered.",
                4,
            )
        })?;
    let credentials = SystemCredentials::new(store.root())?;
    let transport =
        crate::app::backend::Context::load(&store, &registry, &credentials, &instance, deadline)?;
    let token = credentials.get(&instance.credential_key)?;
    let backend = transport.client(&instance, &token, &settings.config)?;
    let session = Session {
        backend: &backend,
        config: &settings.config,
        instance: &instance,
        expected_version: &installation.version,
    };
    session.authorize(conversation.as_ref(), false)?;
    let references = References::open(&store, &credentials, deadline)?;
    let mut source = transport.source(endpoint, &token)?;
    crate::messaging::watch::run_resuming(
        &session,
        &references,
        &mut source,
        sink,
        crate::messaging::watch::Options {
            conversation: conversation.clone(),
            limit: (*limit).into(),
            deadline,
            self_messages: None,
        },
        cancel,
        resume.as_deref(),
    )
}