//! §Fase 34.a (v1.29.0) — Diagnostic anchor for the tools-as-stream-
//! producers cycle.
//!
//! Captures the **current v1.28.0 baseline** for each of the four
//! disjunctions of `produces_stream(F)` defined by the paper §3. Every
//! subsequent sub-fase (34.b-m) inverts a specific aspect of this
//! anchor in lockstep — same forensic-anchor discipline as the
//! 33.a / 33.x.a / 33.y.a / 33.z.a / 33.z.k.a anchors.
//!
//! # The paper's four disjunctions
//!
//! A flow `F` produces a stream iff at least one of:
//!
//! | Disjunct | Adopter shape | v1.28.0 baseline |
//! |---|---|---|
//! | **(a)** Type-level | `step S { output: Stream<T> }` | ✅ live per-chunk via `Backend::stream()` (Fase 33.x.b) |
//! | **(b)** Effect-level (apply syntax) | `step S { apply: <stream-tool> }` | ⚠️ partial — `axon.tool_call` SSE event family active (33.z D5) BUT the tool itself runs synchronously |
//! | **(c)** Effect-level (use_tool syntax) | `use_tool: <stream-tool>` step | ⚠️ partial — same as (b) |
//! | **(d)** Imperative | `perform Stream.Yield(x)` | ✅ static-scan emits one event per static Yield (`bridge_effect_stream_yield`, Fase 33.y.e) |
//!
//! Fase 34's contract: post-34, all 4 disjunctions invoke the SAME
//! `unified_stream_handler` that drains a `Stream<ToolChunk>` through
//! `StreamPolicyEnforcer` with the declared policy. Disjunctions (b)
//! and (c) graduate from "tool runs synchronously" to "tool produces
//! a stream that flows through the enforcer". Disjunctions (a) and
//! (d) stay byte-equal in wire behavior (D4 backwards-compat) but
//! share the same internal handler.
//!
//! # Why anchor BEFORE the lift
//!
//! Each subsequent sub-fase's contract is "invert THIS specific
//! pre-34 assertion". Without an explicit baseline pin, the cycle's
//! progress becomes unfalsifiable — we'd land 34.d "stream-tool
//! produces per-chunk wire" but couldn't verify the inversion
//! happened. The anchor file is the falsifier.
//!
//! # Diagnostic discipline
//!
//! Forensic capture with `eprintln!` (visible under
//! `cargo test -- --nocapture`). Assertions are minimal + defensive:
//! the goal is to PIN the current behavior so post-34 regressions
//! surface as anchor-inversion test failures.
//!
//! # Closed-catalog totality pin
//!
//! §5 asserts the 4 disjunctions are the CLOSED set. The paper §3
//! states `produces_stream(F)` is defined by exactly these four
//! rules; a future axon-lang minor that adds a 5th rule (e.g.
//! "perform Channel.Recv produces a stream") fails this pin. The
//! pin makes catalog growth an explicit code change at the test
//! site — same discipline as 33.z.k.i drift gate.

use axon::axon_server::{build_router, ServerConfig};
use axum::body::Body;
use axum::http::{Request, StatusCode};
use http_body_util::BodyExt;
use tower::ServiceExt;

fn server_cfg() -> ServerConfig {
    ServerConfig {
        host: "127.0.0.1".into(),
        port: 0,
        channel: "memory".into(),
        auth_token: String::new(),
        log_level: "INFO".into(),
        log_format: "json".into(),
        log_file: None,
        database_url: None,
        config_path: None,
        strict_type_driven_transport: false,
        default_backend: None,
        schemas_dir: None,
    }
}

async fn deploy(app: axum::Router, src: &str) -> StatusCode {
    let body = serde_json::json!({
        "source": src,
        "source_file": "fase34_anchor.axon",
        "backend": "stub",
    });
    let req = Request::builder()
        .method("POST")
        .uri("/v1/deploy")
        .header("content-type", "application/json")
        .body(Body::from(body.to_string()))
        .unwrap();
    app.oneshot(req).await.unwrap().status()
}

async fn post_no_accept(app: axum::Router, path: &str) -> (StatusCode, String, String) {
    let req = Request::builder()
        .method("POST")
        .uri(path)
        .header("content-type", "application/json")
        .body(Body::from("{}"))
        .unwrap();
    let resp = app.oneshot(req).await.unwrap();
    let status = resp.status();
    let ct = resp
        .headers()
        .get(axum::http::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("")
        .to_string();
    let bytes = resp.into_body().collect().await.unwrap().to_bytes();
    let body = String::from_utf8_lossy(&bytes).to_string();
    (status, ct, body)
}

fn count_event_lines(body: &str, event_name: &str) -> usize {
    let needle = format!("event: {event_name}");
    body.lines().filter(|l| l.trim_start() == needle).count()
}

fn count_openai_content_chunks(body: &str) -> usize {
    body.lines()
        .filter(|l| {
            l.trim_start().starts_with("data: {")
                && l.contains("\"object\":\"chat.completion.chunk\"")
                && l.contains("\"delta\":{\"content\":")
        })
        .count()
}

// ════════════════════════════════════════════════════════════════════
//  §1 — Disjunct (a) Type-level — `step S { output: Stream<Token> }`
//        baseline: ✅ live per-chunk via Backend::stream()
// ════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn s1_disjunct_a_type_level_stream_emits_axon_dialect_baseline() {
    // Disjunct (a): step declares `output: Stream<Token>`. No tool
    // applied. Q1 Rule 3: type-annotation-only → axon dialect default
    // (W3C-correct baseline). Stub backend emits 1 chunk "(stub)" →
    // 1 axon.token + 1 axon.complete.
    //
    // POST-34: this anchor STAYS byte-identical. The unified handler
    // wraps Backend::stream() chunks in a synthetic Stream<ToolChunk>
    // and drives the SAME handler downstream; wire bytes preserved
    // (D4 byte-compat).
    let src = "flow Chat() -> Unit {\n\
        step Generate { ask: \"hi\" output: Stream<Token> }\n\
    }\n\
    axonendpoint ChatEndpoint { method: POST path: \"/a\" execute: Chat transport: sse }";

    let app = build_router(server_cfg());
    assert_eq!(deploy(app.clone(), src).await, StatusCode::OK);
    let (status, ct, body) = post_no_accept(app, "/a").await;
    assert_eq!(status, StatusCode::OK);
    assert!(ct.contains("text/event-stream"));

    let tokens = count_event_lines(&body, "axon.token");
    let completes = count_event_lines(&body, "axon.complete");

    eprintln!(
        "§1 disjunct (a) anchor (type-level Stream<T> + stub baseline):\n\
         Content-Type: {ct}\n\
         axon.token count = {tokens} (expected 1 — stub emits 1 chunk)\n\
         axon.complete count = {completes} (expected 1 — terminator)\n\
         POST-34 expectation: BYTE-IDENTICAL (unified handler preserves wire shape)\n\
         body sample:\n{}",
        body.chars().take(500).collect::<String>()
    );

    assert_eq!(
        tokens, 1,
        "§1 disjunct (a) v1.28.0: type-level Stream<T> + stub → exactly 1 axon.token"
    );
    assert_eq!(
        completes, 1,
        "§1 disjunct (a) v1.28.0: terminator is axon.complete (axon dialect; type-annotation only → Q1 Rule 3 axon default)"
    );
}

// ════════════════════════════════════════════════════════════════════
//  §2 — Disjunct (b) Effect-level apply syntax —
//        `step S { apply: <stream-tool> }`
//        baseline: ⚠️ PARTIAL — Q1 openai-dialect wire active (33.z.k.g.2)
//        BUT the tool itself runs synchronously (this is the 34.d gap)
// ════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn s2_disjunct_b_apply_syntax_with_stream_effect_emits_openai_baseline() {
    // Disjunct (b): tool declares `effects: <stream:drop_oldest>` +
    // step applies it via `apply:` syntax. Q1 Rule 2: algebraic-effect
    // → openai dialect default (Fase 33.z.k.g.2). Stub backend signals
    // FinishReason::Stop (never ToolUse for stub) so the tool body
    // path doesn't fire today; the LLM upstream emits a single
    // materialized chunk "(stub)" which the openai adapter emits as
    // one content-delta.
    //
    // CRITICAL pre-34 observation: the tool's body is NOT invoked as
    // a stream. The `effects: <stream:drop_oldest>` declaration is
    // captured in the IR but runtime-inert at the tool layer.
    //
    // POST-34 expectation (the 34.d inversion):
    //   1. is_streaming derived from the tool's effect_row
    //   2. dispatcher branches on is_streaming; when true, bypasses
    //      the LLM upstream entirely + invokes tool.stream() directly
    //   3. each tool-internal chunk emerges through the openai
    //      dialect adapter as a separate content-delta frame
    //   4. cancel propagates INTO the tool body (D5 p95 ≤100ms)
    //   5. step_audit captures the tool's chunk count + SHA-256 of
    //      the concatenated tool deltas (D6 audit extension)
    let src = "tool chat_token_stream { description: \"S\" \
               effects: <stream:drop_oldest> }\n\
        flow Chat() -> Unit {\n\
            step Generate { ask: \"hi\" apply: chat_token_stream output: Stream<Token> }\n\
        }\n\
        axonendpoint ChatEndpoint { method: POST path: \"/b\" execute: Chat }";

    let app = build_router(server_cfg());
    assert_eq!(deploy(app.clone(), src).await, StatusCode::OK);
    let (status, ct, body) = post_no_accept(app, "/b").await;
    assert_eq!(status, StatusCode::OK);
    assert!(ct.contains("text/event-stream"));

    let content_chunks = count_openai_content_chunks(&body);
    let has_done = body.contains("data: [DONE]");
    let has_role_marker = body.contains("\"delta\":{\"role\":\"assistant\"}");
    let axon_tokens = count_event_lines(&body, "axon.token");

    eprintln!(
        "§2 disjunct (b) anchor (apply: stream-tool + stub baseline):\n\
         Content-Type: {ct}\n\
         openai content-delta count = {content_chunks} (expected >1 — POST-34.d / \
         36.i the streaming tool body IS invoked as a stream)\n\
         has role marker = {has_role_marker} (expected true — Q1 openai dialect)\n\
         has [DONE] sentinel = {has_done} (expected true — openai terminator)\n\
         axon.token count = {axon_tokens} (expected 0 — Q1 openai dialect; \
         POST-Q5 escape valve would yield axon.token via `transport: sse(axon)`)\n\
         POST-34 expectation: content_chunks ≥ N where N = tool-internal chunk count \
         (NOT just 1 — the tool itself becomes a stream producer)\n\
         body sample:\n{}",
        body.chars().take(800).collect::<String>()
    );

    // §Fase 36.x.e.2 — POST-34.d / 36.i: the tool registry is wired,
    // so `apply: <stream-tool>` routes through the streaming-tool
    // path and the tool body itself produces the stream — MORE than
    // the single pre-34 materialized content-delta. The gap the
    // v1.28.0 comment pre-announced ("POST-34 inverts this") is
    // closed; the test now asserts the inverted (correct) state.
    assert!(
        content_chunks > 1,
        "§2 disjunct (b): the streaming tool body produces a real \
         stream — more than one content-delta reaches the wire \
         (POST-34.d / 36.i). Got {content_chunks}."
    );
    assert!(
        has_done,
        "§2 disjunct (b) v1.28.0: openai dialect terminator [DONE] present"
    );
    assert!(
        has_role_marker,
        "§2 disjunct (b) v1.28.0: openai dialect role marker on first chunk"
    );
    assert_eq!(
        axon_tokens, 0,
        "§2 disjunct (b) v1.28.0: Q1 Rule 2 (algebraic-effect → openai) — \
         no W3C `event: axon.token` lines on the openai-dialect wire"
    );
}

// ════════════════════════════════════════════════════════════════════
//  §3 — Disjunct (c) Effect-level use_tool syntax —
//        `use_tool: <stream-tool>` step
//        baseline: ⚠️ PARTIAL — IRFlowNode::UseTool variant; tool runs
//        synchronously (this is the 34.d gap, same as disjunct (b))
// ════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn s3_disjunct_c_use_tool_syntax_collapses_to_apply_at_runtime() {
    // Disjunct (c): the `use_tool: <name>(args)` syntax produces an
    // `IRFlowNode::UseTool` variant. v1.28.0 baseline: the dispatcher's
    // `use_tool_step` handler (Fase 33.y) treats this as a synchronous
    // tool invocation regardless of whether the tool declares a
    // stream effect.
    //
    // §3 documents that disjunct (c) is, at the dispatcher layer,
    // semantically equivalent to disjunct (b) — both invoke
    // `dispatch_tool_internal` which returns a materialized
    // `ToolResult` synchronously. POST-34: both routes through the
    // SAME `unified_stream_handler` per D3.
    //
    // The use_tool grammar parses, but the integration here is via
    // a flow body that exercises a tool with a stream effect via the
    // canonical apply: path (because the runtime collapse is the same
    // observable behavior + the apply: syntax is the adopter-canonical
    // one tracked by Fase 33.z.k.1's algebraic-effect override).
    //
    // NOTE: this test uses the apply: syntax to exercise the same
    // synchronous-tool path. The dedicated use_tool: parse test lives
    // in axon-frontend/tests/. The wire-byte observation here is
    // semantically equivalent.
    let src = "tool stream_reasoner { description: \"S\" effects: <stream:drop_oldest> }\n\
        flow ReasoningFlow() -> Unit {\n\
            step Reason { ask: \"reason\" apply: stream_reasoner output: Stream<Token> }\n\
        }\n\
        axonendpoint ReasoningEndpoint { method: POST path: \"/c\" execute: ReasoningFlow }";

    let app = build_router(server_cfg());
    assert_eq!(deploy(app.clone(), src).await, StatusCode::OK);
    let (status, ct, body) = post_no_accept(app, "/c").await;
    assert_eq!(status, StatusCode::OK);
    assert!(ct.contains("text/event-stream"));

    let content_chunks = count_openai_content_chunks(&body);

    eprintln!(
        "§3 disjunct (c) anchor (use_tool semantic equivalence to apply baseline):\n\
         Content-Type: {ct}\n\
         openai content-delta count = {content_chunks} (expected >1 — \
         POST-34.d / 36.i both disjuncts (b) + (c) converge through \
         unified_stream_handler invoking tool.stream())"
    );

    // §Fase 36.x.e.2 — `use_tool` collapses to `apply:` at runtime;
    // both route through `unified_stream_handler` invoking the tool's
    // `stream()`. The streaming tool body produces a real multi-chunk
    // stream — the v1.28.0 synchronous-tool gap is closed.
    assert!(
        content_chunks > 1,
        "§3 disjunct (c): `use_tool` collapses to the streaming-tool \
         path — the tool body streams more than one content-delta \
         (POST-34.d / 36.i). Got {content_chunks}."
    );
}

// ════════════════════════════════════════════════════════════════════
//  §4 — Disjunct (d) Imperative — `perform Stream.Yield(x)`
//        baseline: ✅ static-scan emits one event per static Yield
//        (`bridge_effect_stream_yield`, Fase 33.y.e)
// ════════════════════════════════════════════════════════════════════
//
// Disjunct (d) is the algebraic-effect `perform Stream.Yield(x)`
// expression inside a `handle Stream { ... } in { ... }` block.
// Fase 33.y.e shipped `bridge_effect_stream_yield` which statically
// scans the IRPerform tree at flow-compile time + emits one wire
// event per static Yield call site.
//
// The current axon grammar exposes `perform` expressions in
// expression-position contexts (step bodies). Writing a literal
// `perform Stream.Yield(x)` at the source layer requires the
// algebraic-effects surface in the source grammar (Fase 23) which
// is parsed but rarely exercised by adopter shapes.
//
// For the diagnostic anchor, §4 documents the PRE-34 baseline at
// the IR layer + asserts compile-time totality: the static scan
// surface is in `bridge_effect_stream_yield`. The full wire-byte
// integration of disjunct (d) under the unified handler is the
// 34.g milestone.
//
// POST-34 expectation: the static scan's emission path graduates
// into a synthetic `Stream<ToolChunk>` source + the same
// `unified_stream_handler` consumes it. Adopters who write
// `perform Stream.Yield(x)` see byte-identical wire (D4) but the
// internal codepath consolidates.

#[test]
fn s4_disjunct_d_imperative_perform_yield_baseline_pin() {
    // Compile-time pin: the bridge_effect_stream_yield function
    // exists in axon::flow_dispatcher (or wherever 33.y.e wired it).
    // §4 is a STATIC pin — the runtime integration of disjunct (d)
    // doesn't need a separate HTTP fixture because the perform-Yield
    // surface emerges from inside a step body that the dispatcher
    // already exercises via disjuncts (a)/(b)/(c).
    //
    // The closed-catalog discipline: there are EXACTLY 4 disjuncts
    // of produces_stream(F) per the paper §3. §5 below pins this
    // cardinality.
    eprintln!(
        "§4 disjunct (d) anchor (perform Stream.Yield static-scan baseline):\n\
         Pre-34: bridge_effect_stream_yield static-scan emits one wire event \
         per static Yield call site (Fase 33.y.e).\n\
         POST-34: static-scan output graduates to a synthetic Stream<ToolChunk> \
         source feeding the same unified_stream_handler as disjuncts (a/b/c). \
         Wire bytes byte-identical (D4 backwards-compat)."
    );
}

// ════════════════════════════════════════════════════════════════════
//  §5 — Closed-catalog totality pin: 4 disjunctions, no 5th
// ════════════════════════════════════════════════════════════════════

#[test]
fn s5_produces_stream_disjunction_catalog_is_exactly_four() {
    // The paper §3 defines `produces_stream(F)` as the disjunction
    // of exactly 4 rules. Adding a 5th (e.g. "perform Channel.Recv
    // produces a stream") is a deliberate language-level decision
    // that requires a paper update + an explicit sub-fase. Pinning
    // the cardinality here makes any future drift an explicit code
    // change at the test site — same discipline as the 33.z.k.i
    // dialect catalog drift gate.
    const PRODUCES_STREAM_DISJUNCTIONS: &[&str] = &[
        "(a) type-level — step output type is Stream<T>",
        "(b) effect-level apply — step applies tool with <stream:<policy>>",
        "(c) effect-level use_tool — IRFlowNode::UseTool variant",
        "(d) imperative — perform Stream.Yield(x) inside handle Stream",
    ];
    assert_eq!(
        PRODUCES_STREAM_DISJUNCTIONS.len(),
        4,
        "§5 paper §3 totality: produces_stream(F) is the disjunction of \
         EXACTLY 4 rules. Adding a 5th requires a paper update + a \
         deliberate sub-fase + cross-stack drift gate update + adopter \
         docs update."
    );
    eprintln!(
        "§5 anchor: produces_stream disjunctions = {PRODUCES_STREAM_DISJUNCTIONS:?}"
    );
}

// ════════════════════════════════════════════════════════════════════
//  §6 — Tool registry baseline — `is_streaming` does NOT exist yet
// ════════════════════════════════════════════════════════════════════

#[test]
fn s6_tool_registry_is_streaming_field_present_post_34_c() {
    // **Inverted post-34.c** — the pre-34 baseline ("ToolEntry has
    // effect_row but NO is_streaming field") flipped at 34.c.
    // ToolEntry now ships `is_streaming: bool` as a structural
    // field auto-derived from effect_row at registration time via
    // `derive_is_streaming(effect_row)`.
    //
    // The 1-to-1 declaration → runtime contract is pinned by the
    // drift gate `fase34_c_registry_drift.rs` over a synthetic
    // 30-tool corpus.
    use axon::tool_registry::{derive_is_streaming, ToolEntry};

    // Constructing a ToolEntry with a stream effect: is_streaming
    // is set explicitly here (caller responsibility for direct
    // register() path). The register_from_ir() path auto-derives.
    let stream_entry = ToolEntry {
        name: "test_stream_tool".to_string(),
        provider: "stub".to_string(),
        timeout: String::new(),
        runtime: String::new(),
        sandbox: None,
        max_results: None,
        output_schema: String::new(),
        effect_row: vec!["stream:drop_oldest".to_string()],
        parameters: Vec::new(),
        source: axon::tool_registry::ToolSource::Program,
        // §Fase 34.c — Caller sets explicitly OR uses derive helper.
        is_streaming: derive_is_streaming(&["stream:drop_oldest".to_string()]),
    };
    assert!(
        stream_entry.is_streaming,
        "§6 post-34.c: effect_row containing `stream:<policy>` MUST \
         derive is_streaming = true"
    );

    // Non-stream tool: is_streaming should be false.
    let plain_entry = ToolEntry {
        name: "test_plain_tool".to_string(),
        provider: "stub".to_string(),
        timeout: String::new(),
        runtime: String::new(),
        sandbox: None,
        max_results: None,
        output_schema: String::new(),
        effect_row: vec!["compute".to_string(), "read".to_string()],
        parameters: Vec::new(),
        source: axon::tool_registry::ToolSource::Program,
        is_streaming: derive_is_streaming(&[
            "compute".to_string(),
            "read".to_string(),
        ]),
    };
    assert!(
        !plain_entry.is_streaming,
        "§6 post-34.c: effect_row WITHOUT any `stream:` prefix MUST \
         derive is_streaming = false"
    );

    eprintln!(
        "§6 anchor (post-34.c): ToolEntry.is_streaming structural field \
         auto-derived from effect_row.iter().any(|e| e.starts_with(\"stream:\")). \
         The 1-to-1 declaration → runtime contract is the drift gate's \
         load-bearing invariant."
    );
}

// ════════════════════════════════════════════════════════════════════
//  §7 — Dispatcher baseline — Tool::stream() trait does NOT exist
// ════════════════════════════════════════════════════════════════════

#[test]
fn s7_tool_trait_pre_34_has_no_stream_method() {
    // PRE-34 baseline: there is no formal `Tool` trait in axon-rs.
    // Dispatch is a closed match in tool_executor::dispatch over
    // hardcoded tool_name. ToolResult is the single materialized
    // return type.
    //
    // POST-34 (34.b): new `pub trait Tool { execute, stream,
    // is_streaming }` with default impl wrapping execute() as
    // single-chunk stream (D9 backwards-compat: every existing tool
    // keeps working byte-equal via the default impl).
    use axon::tool_executor::ToolResult;

    let result = ToolResult {
        success: true,
        output: "materialized".to_string(),
        tool_name: "test".to_string(),
    };
    assert_eq!(result.success, true);
    assert_eq!(result.output, "materialized");
    eprintln!(
        "§7 anchor: ToolResult is the v1.28.0 single-materialized return type. \
         34.b adds ToolChunk + Tool::stream() default-wraps execute()."
    );
}
