diff --git a/crates/aisix-core/src/models/guardrail.rs b/crates/aisix-core/src/models/guardrail.rs
index dadac78f..c2358a96 100644
--- a/crates/aisix-core/src/models/guardrail.rs
+++ b/crates/aisix-core/src/models/guardrail.rs
@@ -128,6 +128,11 @@ pub struct AzureContentSafetyConfig {
     #[serde(default = "default_acs_timeout_ms")]
     #[schemars(range(max = 4_294_967_295u32))]
     pub timeout_ms: u32,
+    /// Fail-open policy for the output hook. When disabled (the default), an
+    /// Azure outage blocks model output instead of releasing unscanned content.
+    /// The input hook continues to use the top-level `fail_open` policy.
+    #[serde(default)]
+    pub output_fail_open: bool,
 }
 
 fn default_acs_timeout_ms() -> u32 {
@@ -347,6 +352,11 @@ pub struct BedrockConfig {
     pub aws_credentials: BedrockAWSCredentials,
     /// Bedrock guardrail latency policy. Use `timed` with `timeout_ms` to cap wait time.
     pub latency_mode: BedrockLatencyMode,
+    /// Fail-open policy for the output hook. When disabled (the default), a
+    /// Bedrock outage blocks model output instead of releasing unscanned content.
+    /// The input hook continues to use the top-level `fail_open` policy.
+    #[serde(default)]
+    pub output_fail_open: bool,
 }
 
 /// Provider discriminator. The kind drives which `*_config` block is
diff --git a/crates/aisix-guardrails/src/bedrock.rs b/crates/aisix-guardrails/src/bedrock.rs
index c2546960..8319f8e9 100644
--- a/crates/aisix-guardrails/src/bedrock.rs
+++ b/crates/aisix-guardrails/src/bedrock.rs
@@ -7,7 +7,11 @@
 //! so this module only handles plaintext credentials. We never log
 //! the secret.
 //!
-//! Behavior matrix (failure modes):
+//! Behavior matrix (failure modes). The effective `fail_open` is the
+//! outer `Guardrail::fail_open` on the INPUT hook and the independent
+//! `BedrockConfig::output_fail_open` (default fail-closed) on the OUTPUT
+//! hook, so a Bedrock outage can't release unscanned model output by
+//! default:
 //!
 //! | Bedrock response                | `fail_open` | Verdict                        |
 //! |---------------------------------|-------------|--------------------------------|
@@ -56,7 +60,12 @@ pub struct BedrockGuardrail {
     pub guardrail_version: String,
     pub hook_point: GuardrailHookPoint,
     pub latency_mode: BedrockLatencyMode,
+    /// Fail-open policy for the INPUT hook (the outer `Guardrail::fail_open`).
     pub fail_open: bool,
+    /// Fail-open policy for the OUTPUT hook (`BedrockConfig::output_fail_open`,
+    /// default fail-closed). Kept separate so a Bedrock outage can't release
+    /// unscanned model output by default.
+    pub output_fail_open: bool,
     /// AWS SDK client, pre-configured with the row's region and
     /// static credentials. Wrapped in `Arc` so swapping snapshots
     /// doesn't drop a client mid-request.
@@ -149,14 +158,26 @@ impl BedrockGuardrail {
             hook_point,
             latency_mode: cfg.latency_mode.clone(),
             fail_open,
+            output_fail_open: cfg.output_fail_open,
             client: Arc::new(client),
         }
     }
 
+    /// Fail-open policy that governs `source`. The input hook follows the
+    /// outer `fail_open`; the output hook follows `output_fail_open` (default
+    /// fail-closed) so a Bedrock outage can't release unscanned model output.
+    fn fail_open_for(&self, source: &GuardrailContentSource) -> bool {
+        match source {
+            GuardrailContentSource::Output => self.output_fail_open,
+            _ => self.fail_open,
+        }
+    }
+
     /// Run `ApplyGuardrail` against a content block. Wraps the
     /// SDK call with `latency_mode` enforcement and translates the
     /// response/error into a `GuardrailVerdict` per §Behavior matrix.
     async fn apply(&self, source: GuardrailContentSource, text: String) -> GuardrailVerdict {
+        let fail_open = self.fail_open_for(&source);
         let req = self
             .client
             .apply_guardrail()
@@ -203,11 +224,11 @@ impl BedrockGuardrail {
                     GuardrailVerdict::Allow
                 }
             },
-            Err(failure) => self.handle_failure(failure),
+            Err(failure) => self.handle_failure(failure, fail_open),
         }
     }
 
-    fn handle_failure(&self, failure: BedrockFailure) -> GuardrailVerdict {
+    fn handle_failure(&self, failure: BedrockFailure, fail_open: bool) -> GuardrailVerdict {
         let (reason, error_detail, error_source) = failure.log_fields();
         tracing::warn!(
             row = %self.row_name,
@@ -215,10 +236,10 @@ impl BedrockGuardrail {
             failure_tag = reason,
             error = error_detail,
             source = error_source,
-            fail_open = self.fail_open,
+            fail_open = fail_open,
             "bedrock ApplyGuardrail call failed",
         );
-        if self.fail_open {
+        if fail_open {
             GuardrailVerdict::Bypass {
                 reason: reason.into(),
             }
@@ -369,6 +390,8 @@ mod tests {
                 secret_access_key: "TEST".into(),
             },
             latency_mode: BedrockLatencyMode::Serial,
+            // Default fail-closed output (cp-api omits the field when unset).
+            output_fail_open: false,
         }
     }
 
@@ -419,7 +442,7 @@ mod tests {
     #[tokio::test]
     async fn timeout_with_fail_open_true_returns_bypass() {
         let g = build_test(true);
-        let v = g.handle_failure(BedrockFailure::Timeout);
+        let v = g.handle_failure(BedrockFailure::Timeout, g.fail_open);
         match v {
             GuardrailVerdict::Bypass { reason } => assert_eq!(reason, "bedrock_timeout"),
             other => panic!("expected Bypass, got {other:?}"),
@@ -429,14 +452,14 @@ mod tests {
     #[tokio::test]
     async fn timeout_with_fail_open_false_returns_block() {
         let g = build_test(false);
-        let v = g.handle_failure(BedrockFailure::Timeout);
+        let v = g.handle_failure(BedrockFailure::Timeout, g.fail_open);
         assert!(v.is_block(), "expected Block, got {v:?}");
     }
 
     #[tokio::test]
     async fn throttle_with_fail_open_true_tags_throttled() {
         let g = build_test(true);
-        let v = g.handle_failure(BedrockFailure::Throttled);
+        let v = g.handle_failure(BedrockFailure::Throttled, g.fail_open);
         match v {
             GuardrailVerdict::Bypass { reason } => assert_eq!(reason, "bedrock_throttled"),
             other => panic!("expected Bypass, got {other:?}"),
@@ -446,16 +469,72 @@ mod tests {
     #[tokio::test]
     async fn other_5xx_with_fail_open_true_tags_5xx() {
         let g = build_test(true);
-        let v = g.handle_failure(BedrockFailure::Other {
-            detail: "AccessDeniedException(...)".into(),
-            source: None,
-        });
+        let v = g.handle_failure(
+            BedrockFailure::Other {
+                detail: "AccessDeniedException(...)".into(),
+                source: None,
+            },
+            g.fail_open,
+        );
         match v {
             GuardrailVerdict::Bypass { reason } => assert_eq!(reason, "bedrock_5xx"),
             other => panic!("expected Bypass, got {other:?}"),
         }
     }
 
+    /// The OUTPUT hook follows `output_fail_open`, which defaults to
+    /// fail-closed even when the input-side `fail_open` is true. A Bedrock
+    /// outage on the output side must therefore Block, not release unscanned
+    /// model output. This is the P1-3 fix: the single `fail_open` no longer
+    /// governs both hooks.
+    #[tokio::test]
+    async fn output_hook_defaults_fail_closed_even_when_input_fail_open() {
+        // build_test sets input fail_open=true; cfg() leaves output_fail_open
+        // at its serde default (false).
+        let g = build_test(true);
+        assert!(g.fail_open, "input fail_open is true in this fixture");
+        assert!(!g.output_fail_open, "output must default fail-closed");
+        // Input side bypasses (fail_open=true)...
+        assert!(g
+            .handle_failure(
+                BedrockFailure::Timeout,
+                g.fail_open_for(&GuardrailContentSource::Input)
+            )
+            .is_bypass());
+        // ...output side blocks (output_fail_open=false).
+        assert!(g
+            .handle_failure(
+                BedrockFailure::Timeout,
+                g.fail_open_for(&GuardrailContentSource::Output),
+            )
+            .is_block());
+    }
+
+    /// Operators can still opt the output hook back into fail-open by setting
+    /// `output_fail_open: true` — then an outage bypasses on output too. The
+    /// input policy here is the opposite (`fail_open=false`) so the test
+    /// proves the output hook uses its OWN policy, not the input one.
+    #[tokio::test]
+    async fn output_fail_open_true_bypasses_on_output() {
+        let mut c = cfg();
+        c.output_fail_open = true;
+        let g = BedrockGuardrail::new("row", &c, GuardrailHookPoint::Both, false, None);
+        // Output opted into fail-open → bypass.
+        assert!(g
+            .handle_failure(
+                BedrockFailure::Timeout,
+                g.fail_open_for(&GuardrailContentSource::Output),
+            )
+            .is_bypass());
+        // Input still fails closed → block (proves the two policies are split).
+        assert!(g
+            .handle_failure(
+                BedrockFailure::Timeout,
+                g.fail_open_for(&GuardrailContentSource::Input),
+            )
+            .is_block());
+    }
+
     /// Hook-point gating: an Output-only row must allow input checks
     /// without ever hitting AWS. We assert via Allow, never reaching
     /// the apply() codepath.
diff --git a/crates/aisix-guardrails/src/build.rs b/crates/aisix-guardrails/src/build.rs
index 04223d74..cc502c27 100644
--- a/crates/aisix-guardrails/src/build.rs
+++ b/crates/aisix-guardrails/src/build.rs
@@ -124,19 +124,40 @@ fn applied_for(row: &DomainGuardrail) -> AppliedGuardrail {
     }
 }
 
+/// Build the runtime guardrail for a row, applying its `enforcement_mode`.
+/// `block` (the default) returns the guardrail as-is; `monitor` wraps it in
+/// [`MonitorGuardrail`] so it observes violations without blocking. An
+/// unrecognised mode is treated as `block` (fail-safe) with a warning.
 fn build_one(
     row: &DomainGuardrail,
     bedrock_endpoint_url: Option<&str>,
 ) -> Result<Option<Arc<dyn Guardrail>>, BuildError> {
-    // enforcement_mode is not yet implemented: warn so operators don't silently
-    // assume "monitor" means pass-through. See doc comment on the field.
-    if row.enforcement_mode != "block" {
-        tracing::warn!(
-            guardrail_name = %row.name,
-            enforcement_mode = %row.enforcement_mode,
-            "enforcement_mode is not yet implemented; DP will block regardless of this setting",
-        );
+    Ok(build_one_inner(row, bedrock_endpoint_url)?.map(|g| apply_enforcement_mode(row, g)))
+}
+
+/// Wrap `inner` per the row's `enforcement_mode`. See [`build_one`].
+fn apply_enforcement_mode(row: &DomainGuardrail, inner: Arc<dyn Guardrail>) -> Arc<dyn Guardrail> {
+    match row.enforcement_mode.as_str() {
+        "block" => inner,
+        "monitor" => Arc::new(MonitorGuardrail {
+            row_name: row.name.clone(),
+            inner,
+        }),
+        other => {
+            tracing::warn!(
+                guardrail_name = %row.name,
+                enforcement_mode = %other,
+                "unknown enforcement_mode; treating as 'block'",
+            );
+            inner
+        }
     }
+}
+
+fn build_one_inner(
+    row: &DomainGuardrail,
+    bedrock_endpoint_url: Option<&str>,
+) -> Result<Option<Arc<dyn Guardrail>>, BuildError> {
     match &row.config {
         GuardrailKind::Keyword(cfg) => {
             if cfg.patterns.is_empty() {
@@ -268,6 +289,67 @@ enum BuildError {
     FeatureDisabled(&'static str),
 }
 
+/// `enforcement_mode: monitor` decorator. Runs the wrapped guardrail exactly
+/// as configured but never blocks: a `Block` verdict is logged (the operator's
+/// audit signal — "this rule WOULD have blocked") and downgraded to `Allow`.
+/// `Allow` and `Bypass` pass through unchanged.
+///
+/// `runs_on_output` delegates to the inner guardrail so a monitor-mode output
+/// rule still gets its `check_output` called and can record what it observed.
+/// `stream_output_policy` is forced to `EndOfStreamCheck`, though: a guardrail
+/// that can never block must not make the streamed response hold back —
+/// monitor mode observes at end-of-stream without adding hold-back latency,
+/// and it can never weaken a *blocking* peer's hold-back (the chain folds to
+/// the strictest member).
+struct MonitorGuardrail {
+    row_name: String,
+    inner: Arc<dyn Guardrail>,
+}
+
+impl MonitorGuardrail {
+    fn observe(&self, hook: &'static str, verdict: GuardrailVerdict) -> GuardrailVerdict {
+        match verdict {
+            GuardrailVerdict::Block { reason, .. } => {
+                tracing::info!(
+                    guardrail_name = %self.row_name,
+                    hook,
+                    reason = %reason,
+                    "guardrail in monitor mode observed a violation; not blocking (enforcement_mode=monitor)",
+                );
+                GuardrailVerdict::Allow
+            }
+            other => other,
+        }
+    }
+}
+
+#[async_trait]
+impl Guardrail for MonitorGuardrail {
+    fn name(&self) -> &'static str {
+        self.inner.name()
+    }
+
+    fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+
+    async fn check_input(&self, req: &ChatFormat) -> GuardrailVerdict {
+        self.observe("input", self.inner.check_input(req).await)
+    }
+
+    async fn check_output(&self, resp: &ChatResponse) -> GuardrailVerdict {
+        self.observe("output", self.inner.check_output(resp).await)
+    }
+
+    fn stream_output_policy(&self) -> StreamOutputPolicy {
+        StreamOutputPolicy::EndOfStreamCheck
+    }
+
+    fn runs_on_output(&self) -> bool {
+        self.inner.runs_on_output()
+    }
+}
+
 /// Adapter that wraps a snapshot handle and rebuilds the runtime
 /// chain whenever the snapshot pointer changes. The chat handler
 /// holds an `Arc<dyn Guardrail>` pointing at this; it never sees
@@ -661,6 +743,95 @@ mod tests {
         assert!(v.is_block());
     }
 
+    /// P1-3: `enforcement_mode: monitor` observes but never blocks. The same
+    /// keyword rule that blocks under the default `block` mode must Allow the
+    /// matching input when the row is in monitor mode — operators get the
+    /// audit log without the request being rejected.
+    #[tokio::test]
+    async fn monitor_mode_observes_but_does_not_block() {
+        let table: ResourceTable<DomainGuardrail> = ResourceTable::default();
+        table.insert(entry(
+            "watch-secrets",
+            "g-1",
+            parse(
+                r#"{
+                    "name": "watch-secrets",
+                    "enforcement_mode": "monitor",
+                    "kind": "keyword",
+                    "patterns": [{ "kind": "literal", "value": "AKIA" }]
+                }"#,
+            ),
+        ));
+        let chain = build_chain_from_snapshot(&table, None);
+        assert_eq!(chain.len(), 1, "monitor-mode row still materialises");
+        // Would block under `block` mode; monitor downgrades to Allow.
+        let v = chain.check_input(&req("here is AKIAEXAMPLE")).await;
+        assert!(!v.is_block(), "monitor mode must not block, got {v:?}",);
+        assert_eq!(v, GuardrailVerdict::Allow);
+        // Output hook is monitored the same way.
+        let resp = ChatResponse {
+            id: "r".into(),
+            model: "m".into(),
+            message: ChatMessage::assistant("leaking AKIAEXAMPLE"),
+            finish_reason: aisix_gateway::FinishReason::Stop,
+            usage: aisix_gateway::UsageStats::new(0, 0),
+        };
+        assert!(!chain.check_output(&resp).await.is_block());
+    }
+
+    /// A monitor-mode guardrail must not force streamed output to hold back —
+    /// it can never block, so hold-back would be pure latency. It folds to the
+    /// no-hold-back policy (and, in a mixed chain, can't weaken a blocking
+    /// peer because the chain keeps the strictest member's policy).
+    #[tokio::test]
+    async fn monitor_mode_does_not_force_stream_holdback() {
+        let table: ResourceTable<DomainGuardrail> = ResourceTable::default();
+        table.insert(entry(
+            "watch-out",
+            "g-1",
+            parse(
+                r#"{
+                    "name": "watch-out",
+                    "enforcement_mode": "monitor",
+                    "kind": "keyword",
+                    "hook_point": "output",
+                    "patterns": [{ "kind": "literal", "value": "secret" }]
+                }"#,
+            ),
+        ));
+        let chain = build_chain_from_snapshot(&table, None);
+        assert!(
+            !chain.stream_output_policy().holds_back(),
+            "monitor-mode output rule must not hold the stream back",
+        );
+    }
+
+    /// An unrecognised enforcement_mode is treated as `block` (fail-safe).
+    #[tokio::test]
+    async fn unknown_enforcement_mode_falls_back_to_block() {
+        let table: ResourceTable<DomainGuardrail> = ResourceTable::default();
+        table.insert(entry(
+            "g",
+            "g-1",
+            parse(
+                r#"{
+                    "name": "g",
+                    "enforcement_mode": "audit-only-typo",
+                    "kind": "keyword",
+                    "patterns": [{ "kind": "literal", "value": "AKIA" }]
+                }"#,
+            ),
+        ));
+        let chain = build_chain_from_snapshot(&table, None);
+        assert!(
+            chain
+                .check_input(&req("here is AKIAEXAMPLE"))
+                .await
+                .is_block(),
+            "unknown mode must default to block, not silently pass through",
+        );
+    }
+
     #[tokio::test]
     async fn disabled_row_is_dropped() {
         let table: ResourceTable<DomainGuardrail> = ResourceTable::default();
diff --git a/crates/aisix-guardrails/src/prompt_shield.rs b/crates/aisix-guardrails/src/prompt_shield.rs
index 67fea702..432a3b97 100644
--- a/crates/aisix-guardrails/src/prompt_shield.rs
+++ b/crates/aisix-guardrails/src/prompt_shield.rs
@@ -23,7 +23,11 @@
 //! The cp-api decrypts the envelope-encrypted `api_key` at kine-projection
 //! time so this module only handles plaintext keys. The key is never logged.
 //!
-//! Behavior matrix (failure modes):
+//! Behavior matrix (failure modes). The effective `fail_open` is the outer
+//! `Guardrail::fail_open` on the INPUT hook and the independent
+//! `AzureContentSafetyConfig::output_fail_open` (default fail-closed) on the
+//! OUTPUT hook, so an Azure outage can't release unscanned model output by
+//! default:
 //!
 //! | API response                    | `fail_open` | Verdict                               |
 //! |---------------------------------|-------------|---------------------------------------|
@@ -69,7 +73,12 @@ pub struct PromptShieldGuardrail {
     /// Plaintext subscription key (decrypted by cp-api before kine write).
     api_key: String,
     pub(crate) hook_point: GuardrailHookPoint,
+    /// Fail-open policy for the INPUT hook (the outer `Guardrail::fail_open`).
     fail_open: bool,
+    /// Fail-open policy for the OUTPUT hook (`AzureContentSafetyConfig::
+    /// output_fail_open`, default fail-closed). Kept separate so an Azure
+    /// outage can't release unscanned model output by default.
+    output_fail_open: bool,
     /// Call timeout. 0 ms in config → `Duration::ZERO` here, which
     /// `tokio::time::timeout` treats as "already elapsed"; callers that
     /// want no timeout should pass a very large value instead.
@@ -103,6 +112,7 @@ impl PromptShieldGuardrail {
             api_key: cfg.api_key.clone(),
             hook_point,
             fail_open,
+            output_fail_open: cfg.output_fail_open,
             timeout: Duration::from_millis(cfg.timeout_ms as u64),
             client: Arc::new(client),
         }
@@ -111,7 +121,7 @@ impl PromptShieldGuardrail {
     /// Check `text` against Prompt Shield, splitting it into ≤10 000-char
     /// chunks. Returns `Block` on the first chunk where
     /// `attackDetected=true`; returns `Allow` when all chunks pass.
-    async fn shield(&self, text: &str) -> GuardrailVerdict {
+    async fn shield(&self, text: &str, fail_open: bool) -> GuardrailVerdict {
         for chunk in chunk_text(text, MAX_PROMPT_CHARS) {
             match self.call_api(&chunk).await {
                 Ok(true) => {
@@ -121,7 +131,7 @@ impl PromptShieldGuardrail {
                     ));
                 }
                 Ok(false) => {} // clean — continue to next chunk
-                Err(failure) => return self.handle_failure(failure),
+                Err(failure) => return self.handle_failure(failure, fail_open),
             }
         }
         GuardrailVerdict::Allow
@@ -178,7 +188,7 @@ impl PromptShieldGuardrail {
         Ok(attacked)
     }
 
-    fn handle_failure(&self, failure: AcsFailure) -> GuardrailVerdict {
+    fn handle_failure(&self, failure: AcsFailure, fail_open: bool) -> GuardrailVerdict {
         let tag = failure.bypass_tag();
         // ConfigError is already logged at error level in call_api(); skip
         // the generic warn here so operators see exactly one log line per
@@ -187,11 +197,11 @@ impl PromptShieldGuardrail {
             tracing::warn!(
                 row = %self.row_name,
                 failure = ?failure,
-                fail_open = self.fail_open,
+                fail_open = fail_open,
                 "azure content safety call failed",
             );
         }
-        if self.fail_open {
+        if fail_open {
             GuardrailVerdict::Bypass { reason: tag.into() }
         } else {
             GuardrailVerdict::block(format!("azure content safety unavailable ({tag})"))
@@ -283,7 +293,7 @@ impl Guardrail for PromptShieldGuardrail {
         if text.is_empty() {
             return GuardrailVerdict::Allow;
         }
-        self.shield(&text).await
+        self.shield(&text, self.fail_open).await
     }
 
     async fn check_output(&self, resp: &ChatResponse) -> GuardrailVerdict {
@@ -297,7 +307,9 @@ impl Guardrail for PromptShieldGuardrail {
         if text.is_empty() {
             return GuardrailVerdict::Allow;
         }
-        self.shield(&text).await
+        // Output hook follows its own fail policy (default fail-closed) so an
+        // Azure outage can't release unscanned model output.
+        self.shield(&text, self.output_fail_open).await
     }
 }
 
@@ -382,6 +394,8 @@ mod tests {
             endpoint: endpoint.to_owned(),
             api_key: "test-key-abc".to_owned(),
             timeout_ms: 5_000,
+            // Default fail-closed output (cp-api omits the field when unset).
+            output_fail_open: false,
         }
     }
 
@@ -499,7 +513,7 @@ mod tests {
     #[tokio::test]
     async fn timeout_fail_open_true_returns_bypass() {
         let g = build("http://unused", true);
-        let v = g.handle_failure(AcsFailure::Timeout);
+        let v = g.handle_failure(AcsFailure::Timeout, g.fail_open);
         match v {
             GuardrailVerdict::Bypass { reason } => assert_eq!(reason, "azure_cs_timeout"),
             other => panic!("expected Bypass, got {other:?}"),
@@ -509,20 +523,83 @@ mod tests {
     #[tokio::test]
     async fn timeout_fail_open_false_returns_block() {
         let g = build("http://unused", false);
-        let v = g.handle_failure(AcsFailure::Timeout);
+        let v = g.handle_failure(AcsFailure::Timeout, g.fail_open);
         assert!(v.is_block(), "expected Block, got {v:?}");
     }
 
     #[tokio::test]
     async fn throttled_fail_open_true_returns_bypass_throttled() {
         let g = build("http://unused", true);
-        let v = g.handle_failure(AcsFailure::Throttled);
+        let v = g.handle_failure(AcsFailure::Throttled, g.fail_open);
         match v {
             GuardrailVerdict::Bypass { reason } => assert_eq!(reason, "azure_cs_throttled"),
             other => panic!("expected Bypass, got {other:?}"),
         }
     }
 
+    /// P1-3: the OUTPUT hook follows `output_fail_open`, which defaults to
+    /// fail-closed even when the input-side `fail_open` is true. An Azure
+    /// outage on the output side must Block, not release unscanned output.
+    #[tokio::test]
+    async fn output_defaults_fail_closed_even_when_input_fail_open() {
+        let g = build("http://unused", true);
+        assert!(g.fail_open, "input fail_open is true in this fixture");
+        assert!(!g.output_fail_open, "output must default fail-closed");
+        // Input policy bypasses, output policy blocks.
+        assert!(g
+            .handle_failure(AcsFailure::Timeout, g.fail_open)
+            .is_bypass());
+        assert!(g
+            .handle_failure(AcsFailure::Timeout, g.output_fail_open)
+            .is_block());
+    }
+
+    /// A 5xx on the output hook fails closed by default — exercised through
+    /// the real HTTP path so the wiring (check_output → shield →
+    /// handle_failure) is covered end-to-end, not just the mapping fn. The
+    /// matcher pins the shield path + version and `expect(1)`, so the verdict
+    /// can't come from an unmatched-route 404 (which would also block).
+    #[tokio::test]
+    async fn output_5xx_fails_closed_by_default() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/contentsafety/text:shieldPrompt"))
+            .and(query_param("api-version", "2024-09-01"))
+            .respond_with(ResponseTemplate::new(500))
+            .expect(1)
+            .mount(&server)
+            .await;
+        // input fail_open=true; output_fail_open defaults false.
+        let g = build(&server.uri(), true);
+        assert!(
+            g.check_output(&resp("model output")).await.is_block(),
+            "output hook must fail closed on Azure 5xx by default",
+        );
+    }
+
+    /// Operators can opt the output hook back into fail-open. Driven through
+    /// the real HTTP path with the input policy set the OPPOSITE way
+    /// (`fail_open=false`), so a Bypass proves the output hook follows
+    /// `output_fail_open`, not the input policy.
+    #[tokio::test]
+    async fn output_fail_open_true_bypasses_on_output() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/contentsafety/text:shieldPrompt"))
+            .and(query_param("api-version", "2024-09-01"))
+            .respond_with(ResponseTemplate::new(500))
+            .expect(1)
+            .mount(&server)
+            .await;
+        let mut c = cfg(&server.uri());
+        c.output_fail_open = true;
+        let g = PromptShieldGuardrail::new("row", &c, GuardrailHookPoint::Both, false);
+        match g.check_output(&resp("model output")).await {
+            GuardrailVerdict::Bypass { reason } => assert_eq!(reason, "azure_cs_5xx"),
+            other => panic!("expected Bypass(azure_cs_5xx), got {other:?}"),
+        }
+    }
+
     // -----------------------------------------------------------------------
     // Hook-point gating (no HTTP needed)
     // -----------------------------------------------------------------------
diff --git a/schemas/resources/guardrail.schema.json b/schemas/resources/guardrail.schema.json
index 129ee8ee..08faab7c 100644
--- a/schemas/resources/guardrail.schema.json
+++ b/schemas/resources/guardrail.schema.json
@@ -212,6 +212,11 @@
           ],
           "description": "Bedrock guardrail latency policy. Use `timed` with `timeout_ms` to cap wait time."
         },
+        "output_fail_open": {
+          "default": false,
+          "description": "Fail-open policy for the output hook. When disabled (the default), a Bedrock outage blocks model output instead of releasing unscanned content. The input hook continues to use the top-level `fail_open` policy.",
+          "type": "boolean"
+        },
         "region": {
           "description": "AWS region for the Bedrock endpoint, such as `us-east-1`.",
           "minLength": 1,
@@ -247,6 +252,11 @@
           ],
           "type": "string"
         },
+        "output_fail_open": {
+          "default": false,
+          "description": "Fail-open policy for the output hook. When disabled (the default), an Azure outage blocks model output instead of releasing unscanned content. The input hook continues to use the top-level `fail_open` policy.",
+          "type": "boolean"
+        },
         "timeout_ms": {
           "default": 5000,
           "description": "HTTP call timeout in milliseconds. A value of `0` triggers the timeout immediately.",
diff --git a/tests/e2e/src/cases/guardrail-monitor-mode-e2e.test.ts b/tests/e2e/src/cases/guardrail-monitor-mode-e2e.test.ts
new file mode 100644
index 00000000..4456786b
--- /dev/null
+++ b/tests/e2e/src/cases/guardrail-monitor-mode-e2e.test.ts
@@ -0,0 +1,134 @@
+import { createHash } from "node:crypto";
+import OpenAI, { APIError } from "openai";
+import { afterAll, beforeAll, describe, expect, test } from "vitest";
+import {
+  AdminClient,
+  EtcdClient,
+  spawnApp,
+  startOpenAiUpstream,
+  waitConfigPropagation,
+  type OpenAiUpstream,
+  type SpawnedApp,
+} from "../harness/index.js";
+
+// E2E: `enforcement_mode: "monitor"` observes a guardrail violation WITHOUT
+// blocking the request (issue 788 P1-3). The same keyword rule that 422s in
+// the default "block" mode must let the forbidden-word request through —
+// 200 + a real upstream hit — once flipped to monitor.
+//
+// The block→monitor flip is what makes this test non-racy: step 1 proves the
+// rule is loaded and actively blocking, so step 3's transition to 200 can
+// only mean monitor mode took effect (not "guardrail not loaded yet").
+
+const CALLER_PLAINTEXT = "sk-gr-monitor-caller";
+const CALLER_KEY_HASH = createHash("sha256")
+  .update(CALLER_PLAINTEXT)
+  .digest("hex");
+
+const FORBIDDEN_WORD = "supersecret";
+
+function guardrailBody(enforcementMode: "block" | "monitor") {
+  return {
+    name: "gr-monitor-e2e",
+    enabled: true,
+    hook_point: "input",
+    enforcement_mode: enforcementMode,
+    kind: "keyword",
+    patterns: [{ kind: "literal", value: FORBIDDEN_WORD }],
+  };
+}
+
+describe("guardrail e2e: enforcement_mode monitor observes without blocking", () => {
+  let app: SpawnedApp | undefined;
+  let upstream: OpenAiUpstream | undefined;
+  let admin: AdminClient | undefined;
+  let etcdReachable = false;
+  let guardrailId: string | undefined;
+
+  beforeAll(async () => {
+    etcdReachable = await new EtcdClient().ping();
+    if (!etcdReachable) return;
+
+    upstream = await startOpenAiUpstream();
+    app = await spawnApp();
+    admin = new AdminClient(app.adminUrl, app.adminKey);
+
+    const pk = await admin.createProviderKey({
+      display_name: "gr-monitor-pk",
+      secret: "sk-mock",
+      api_base: `${upstream.baseUrl}/v1`,
+    });
+    await admin.createModel({
+      display_name: "gr-monitor-e2e",
+      provider: "openai",
+      model_name: "gpt-4o-mini",
+      provider_key_id: pk.id,
+    });
+    await admin.createApiKey({
+      key_hash: CALLER_KEY_HASH,
+      allowed_models: ["gr-monitor-e2e"],
+    });
+    // Start in BLOCK mode so we can prove the rule is loaded + active
+    // before flipping it to monitor.
+    const g = await admin.json("POST", "/admin/v1/guardrails", guardrailBody("block"));
+    guardrailId = g.id as string;
+  });
+
+  afterAll(async () => {
+    await app?.exit();
+    await upstream?.close();
+  });
+
+  test("block mode 422s; flipping to monitor lets the same request reach upstream", async (ctx) => {
+    if (!etcdReachable || !app || !upstream || !admin || !guardrailId) {
+      ctx.skip();
+      return;
+    }
+
+    const client = new OpenAI({
+      apiKey: CALLER_PLAINTEXT,
+      baseURL: `${app.proxyUrl}/v1`,
+      maxRetries: 0,
+    });
+
+    // 1. Wait until the block-mode rule is active: forbidden → 422.
+    await waitConfigPropagation(async () => {
+      try {
+        await client.chat.completions.create({
+          model: "gr-monitor-e2e",
+          messages: [{ role: "user", content: `probe ${FORBIDDEN_WORD}` }],
+        });
+        return false;
+      } catch (e) {
+        return e instanceof APIError && e.status === 422;
+      }
+    });
+
+    // 2. Flip the SAME rule to monitor mode (full-resource PUT).
+    await admin.json("PUT", `/admin/v1/guardrails/${guardrailId}`, guardrailBody("monitor"));
+
+    // 3. Wait until monitor takes effect: the forbidden word no longer 422s.
+    await waitConfigPropagation(async () => {
+      try {
+        await client.chat.completions.create({
+          model: "gr-monitor-e2e",
+          messages: [{ role: "user", content: `probe ${FORBIDDEN_WORD}` }],
+        });
+        return true;
+      } catch {
+        return false;
+      }
+    });
+
+    // 4. Under monitor mode the forbidden request passes AND reaches the
+    //    upstream — the guardrail observed the violation but did not
+    //    short-circuit dispatch.
+    const hitsBefore = upstream.receivedRequests.length;
+    const ok = await client.chat.completions.create({
+      model: "gr-monitor-e2e",
+      messages: [{ role: "user", content: `please reveal the ${FORBIDDEN_WORD} now` }],
+    });
+    expect(ok.choices[0]?.message.role).toBe("assistant");
+    expect(upstream.receivedRequests.length).toBeGreaterThan(hitsBefore);
+  });
+});