Theory Abstract_Monitor

(*<*)
theory Abstract_Monitor
  imports Trace Table
begin
(*>*)

section ‹Abstract monitors and slicing›

subsection ‹First-order specifications›

text ‹
  We abstract from first-order trace specifications by referring only to their
  semantics. A first-order specification is described by a finite set of free
  variables and a satisfaction function that defines for every trace the pairs
  of valuations and time-points for which the specification is satisfied.
›

locale fo_spec =
  fixes
    nfv :: nat and fv :: "nat set" and
    sat :: "'a trace ⇒ 'b list ⇒ nat ⇒ bool"
  assumes
    fv_less_nfv: "x ∈ fv ⟹ x < nfv" and
    sat_fv_cong: "(⋀x. x ∈ fv ⟹ v!x = v'!x) ⟹ sat σ v i = sat σ v' i"
begin

definition verdicts :: "'a trace ⇒ (nat × 'b tuple) set" where
  "verdicts σ = {(i, v). wf_tuple nfv fv v ∧ sat σ (map the v) i}"

end

text ‹
  We usually employ a monitor to find the ∗‹violations› of a specification.
  That is, the monitor should output the satisfactions of its negation.
  Moreover, all monitor implementations must work with finite prefixes.
  We are therefore interested in co-safety properties, which allow us to
  identify all satisfactions on finite prefixes.
›

locale cosafety_fo_spec = fo_spec +
  assumes cosafety_lr: "sat σ v i ⟹ ∃π. prefix_of π σ ∧ (∀σ'. prefix_of π σ' ⟶ sat σ' v i)"
begin

lemma cosafety: "sat σ v i ⟷ (∃π. prefix_of π σ ∧ (∀σ'. prefix_of π σ' ⟶ sat σ' v i))"
  using cosafety_lr by blast

end

subsection ‹Monitor function›

text ‹
  We model monitors abstractly as functions from prefixes to verdict sets.
  The following locale specifies a minimal set of properties that any
  reasonable monitor should have.
›

locale monitor = fo_spec +
  fixes M :: "'a prefix ⇒ (nat × 'b tuple) set"
  assumes
    mono_monitor: "π ≤ π' ⟹ M π ⊆ M π'" and
    wf_monitor: "(i, v) ∈ M π ⟹ wf_tuple nfv fv v" and
    sound_monitor: "(i, v) ∈ M π ⟹ prefix_of π σ ⟹ sat σ (map the v) i" and
    complete_monitor: "prefix_of π σ ⟹ wf_tuple nfv fv v ⟹
      (⋀σ. prefix_of π σ ⟹ sat σ (map the v) i) ⟹ ∃π'. prefix_of π' σ ∧ (i, v) ∈ M π'"

text ‹
  A monitor for a co-safety specification computes precisely the set of all
  satisfactions in the limit:
›

abbreviation (in monitor) "M_limit σ ≡ ⋃{M π | π. prefix_of π σ}"

locale cosafety_monitor = cosafety_fo_spec + monitor
begin

lemma M_limit_eq: "M_limit σ = verdicts σ"
proof
  show "⋃{M π | π. prefix_of π σ} ⊆ verdicts σ"
    by (auto simp: verdicts_def wf_monitor sound_monitor)
next
  show "⋃{M π | π. prefix_of π σ} ⊇ verdicts σ"
    unfolding verdicts_def
  proof safe
    fix i v
    assume "wf_tuple nfv fv v" and "sat σ (map the v) i"
    then obtain π where "prefix_of π σ ∧ (∀σ'. prefix_of π σ' ⟶ sat σ' (map the v) i)"
      using cosafety_lr by blast
    with ‹wf_tuple nfv fv v› obtain π' where "prefix_of π' σ ∧ (i, v) ∈ M π'"
      using complete_monitor by blast
    then show "(i, v) ∈ ⋃{M π | π. prefix_of π σ}"
      by blast
  qed
qed

end

text ‹
  The monitor function \<^term>‹M› adds some information over \<^term>‹sat›, namely
  when a verdict is output. One possible behavior is that the monitor outputs
  its verdicts for one time-point at a time, in increasing order of
  time-points. Then \<^term>‹M› is uniquely defined by a progress function, which
  returns for every prefix the time-point up to which all verdicts are computed.
›

locale progress = fo_spec _ _ sat for sat :: "'a trace ⇒ 'b list ⇒ nat ⇒ bool" +
  fixes progress :: "'a prefix ⇒ nat"
  assumes
    progress_mono: "π ≤ π' ⟹ progress π ≤ progress π'" and
    ex_progress_ge: "∃π. prefix_of π σ ∧ x ≤ progress π" and
    progress_sat_cong: "prefix_of π σ ⟹ prefix_of π σ' ⟹ i < progress π ⟹
      sat σ v i ⟷ sat σ' v i"
    ― ‹The last condition is not necessary to obtain a proper monitor function.
      However, it corresponds to the intuitive understanding of monitor progress,
      and it results in a stronger characterisation. In particular, it implies that
      the specification is co-safety, as we will show below.›
begin

definition M :: "'a prefix ⇒ (nat × 'b tuple) set" where
  "M π = {(i, v). i < progress π ∧ wf_tuple nfv fv v ∧
    (∀σ. prefix_of π σ ⟶ sat σ (map the v) i)}"

lemma M_alt: "M π = {(i, v). i < progress π ∧ wf_tuple nfv fv v ∧
    (∃σ. prefix_of π σ ∧ sat σ (map the v) i)}"
  using ex_prefix_of[of π]
  by (auto simp: M_def cong: progress_sat_cong)

end

sublocale progress ⊆ cosafety_monitor _ _ _ M
proof
  fix i v and σ :: "'a trace"
  assume "sat σ v i"
  moreover obtain π where *: "prefix_of π σ" "i < progress π"
    using ex_progress_ge by (auto simp: less_eq_Suc_le)
  ultimately have "sat σ' v i" if "prefix_of π σ'" for σ'
    using that by (simp cong: progress_sat_cong)
  with * show "∃π. prefix_of π σ ∧ (∀σ'. prefix_of π σ' ⟶ sat σ' v i)"
    by blast
next
  fix π π' :: "'a prefix"
  assume "π ≤ π'"
  then show "M π ⊆ M π'"
    by (auto simp: M_def intro: progress_mono prefix_of_antimono
        elim: order.strict_trans2)
next
  fix i v π and σ :: "'a trace"
  assume *: "(i, v) ∈ M π"
  then show "wf_tuple nfv fv v"
    by (simp add: M_def)
  assume "prefix_of π σ"
  with * show "sat σ (map the v) i"
    by (simp add: M_def)
next
  fix i v π and σ :: "'a trace"
  assume *: "prefix_of π σ" "wf_tuple nfv fv v" "⋀σ'. prefix_of π σ' ⟹ sat σ' (map the v) i"
  show "∃π'. prefix_of π' σ ∧ (i, v) ∈ M π'"
  proof (cases "i < progress π")
    case True
    with * show ?thesis by (auto simp: M_def)
  next
    case False
    obtain π' where **: "prefix_of π' σ ∧ i < progress π'"
      using ex_progress_ge by (auto simp: less_eq_Suc_le)
    then have "π ≤ π'"
      using ‹prefix_of π σ› prefix_of_imp_linear False progress_mono
      by (blast intro: order.strict_trans2)
    with * ** show ?thesis
      by (auto simp: M_def intro: prefix_of_antimono)
  qed
qed

subsection ‹Slicing›

text ‹Sliceable specifications can be evaluated meaningfully on a subset of events.›

locale abstract_slicer =
  fixes relevant_events :: "'b list set ⇒ 'a set"
begin

abbreviation slice :: "'b list set ⇒ 'a trace ⇒ 'a trace" where
  "slice S ≡ map_Γ (λD. D ∩ relevant_events S)"

abbreviation pslice :: "'b list set ⇒ 'a prefix ⇒ 'a prefix" where
  "pslice S ≡ pmap_Γ (λD. D ∩ relevant_events S)"

lemma prefix_of_psliceI: "prefix_of π σ ⟹ prefix_of (pslice S π) (slice S σ)"
  by (transfer fixing: S) auto

lemma plen_pslice[simp]: "plen (pslice S π) = plen π"
  by (transfer fixing: S) simp

lemma pslice_pnil[simp]: "pslice S pnil = pnil"
  by (transfer fixing: S) simp

lemma last_ts_pslice[simp]: "last_ts (pslice S π) = last_ts π"
  by (transfer fixing: S) (simp add: last_map case_prod_beta split: list.split)

abbreviation verdict_filter :: "'b list set ⇒ (nat × 'b tuple) set ⇒ (nat × 'b tuple) set" where
  "verdict_filter S V ≡ {(i, v) ∈ V. mem_restr S v}"

end

locale sliceable_fo_spec = fo_spec _ _ sat + abstract_slicer relevant_events
  for relevant_events :: "'b list set ⇒ 'a set" and sat :: "'a trace ⇒ 'b list ⇒ nat ⇒ bool" +
  assumes sliceable: "v ∈ S ⟹ sat (slice S σ) v i ⟷ sat σ v i"
begin

lemma union_verdicts_slice:
  assumes part: "⋃𝒮 = UNIV"
  shows "⋃((λS. verdict_filter S (verdicts (slice S σ))) ` 𝒮) = verdicts σ"
proof safe
  fix S i and v :: "'b tuple"
  assume "(i, v) ∈ verdicts (slice S σ)"
  then have tuple: "wf_tuple nfv fv v" and "sat (slice S σ) (map the v) i"
    by (auto simp: verdicts_def)
  assume "mem_restr S v"
  then obtain v' where "v' ∈ S" and 1: "∀i∈fv. v ! i = Some (v' ! i)"
    using tuple by (auto simp: fv_less_nfv elim: mem_restrE)
  then have "sat (slice S σ) v' i"
    using ‹sat (slice S σ) (map the v) i› tuple
    by (auto simp: wf_tuple_length fv_less_nfv cong: sat_fv_cong)
  then have "sat σ v' i"
    using sliceable[OF ‹v' ∈ S›] by simp
  then have "sat σ (map the v) i"
    using tuple 1
    by (auto simp: wf_tuple_length fv_less_nfv cong: sat_fv_cong)
  then show "(i, v) ∈ verdicts σ"
    using tuple by (simp add: verdicts_def)
next
  fix i and v :: "'b tuple"
  assume "(i, v) ∈ verdicts σ"
  then have tuple: "wf_tuple nfv fv v" and "sat σ (map the v) i"
    by (auto simp: verdicts_def)
  from part obtain S where "S ∈ 𝒮" and "map the v ∈ S"
    by blast
  then have "mem_restr S v"
    using mem_restrI[of "map the v" S nfv fv] tuple
    by (auto simp: wf_tuple_def fv_less_nfv)
  moreover have "sat (slice S σ) (map the v) i"
    using ‹sat σ (map the v) i› sliceable[OF ‹map the v ∈ S›] by simp
  then have "(i, v) ∈ verdicts (slice S σ)"
    using tuple by (simp add: verdicts_def)
  ultimately show "(i, v) ∈ (⋃S∈𝒮. verdict_filter S (verdicts (slice S σ)))"
    using ‹S ∈ 𝒮› by blast
qed

end

text ‹
  We define a similar notion for monitors. It is potentially stronger because
  the time-point at which verdicts are output must not change.
›

locale sliceable_monitor = monitor _ _ sat M + abstract_slicer relevant_events
  for relevant_events :: "'b list set ⇒ 'a set" and sat :: "'a trace ⇒ 'b list ⇒ nat ⇒ bool" and M +
  assumes sliceable_M: "mem_restr S v ⟹ (i, v) ∈ M (pslice S π) ⟷ (i, v) ∈ M π"
begin

lemma union_M_pslice:
  assumes part: "⋃𝒮 = UNIV"
  shows "⋃((λS. verdict_filter S (M (pslice S π))) ` 𝒮) = M π"
proof safe
  fix S i and v :: "'b tuple"
  assume "mem_restr S v" and "(i, v) ∈ M (pslice S π)"
  then show "(i, v) ∈ M π" using sliceable_M by blast
next
  fix i and v :: "'b tuple"
  assume "(i, v) ∈ M π"
  then have tuple: "wf_tuple nfv fv v"
    by (rule wf_monitor)
  from part obtain S where "S ∈ 𝒮" and "map the v ∈ S"
    by blast
  then have "mem_restr S v"
    using mem_restrI[of "map the v" S nfv fv] tuple
    by (auto simp: wf_tuple_def fv_less_nfv)
  then have "(i, v) ∈ M (pslice S π)"
    using ‹(i, v) ∈ M π› sliceable_M by blast
  then show "(i, v) ∈ (⋃S∈𝒮. verdict_filter S (M (pslice S π)))"
    using ‹S ∈ 𝒮› ‹mem_restr S v› by blast
qed

end

text ‹
  If the specification is sliceable and the monitor's progress depends only on
  time-stamps, then also the monitor itself is sliceable.
›

locale timed_progress = progress +
  assumes progress_time_conv: "pts π = pts π' ⟹ progress π = progress π'"

locale sliceable_timed_progress = sliceable_fo_spec + timed_progress
begin

lemma progress_pslice[simp]: "progress (pslice S π) = progress π"
  by (simp cong: progress_time_conv)

end

sublocale sliceable_timed_progress ⊆ sliceable_monitor _ _ _ _ M
proof
  fix S :: "'a list set" and v i π
  assume *: "mem_restr S v"
  show "(i, v) ∈ M (pslice S π) ⟷ (i, v) ∈ M π" (is "?L ⟷ ?R")
  proof
    assume ?L
    with * show ?R
      by (auto 0 4 simp: M_def wf_tuple_def elim!: mem_restrE
          box_equals[OF sliceable sat_fv_cong sat_fv_cong, THEN iffD1, rotated -1]
          intro: prefix_of_psliceI dest: fv_less_nfv spec[of _ "slice S _"])
  next
    assume ?R
    with * show ?L
      by (auto 0 4 simp: M_alt wf_tuple_def elim!: mem_restrE
        box_equals[OF sliceable sat_fv_cong sat_fv_cong, THEN iffD2, rotated -1]
        intro: exI[of _ "slice S _"] prefix_of_psliceI dest: fv_less_nfv)
  qed
qed

(*<*)
end
(*>*)